diff options
Diffstat (limited to 'subagents')
| -rw-r--r-- | subagents/frontmatter.lua | 96 | ||||
| -rw-r--r-- | subagents/jobs.lua | 427 | ||||
| -rw-r--r-- | subagents/luatool.lua | 193 | ||||
| -rw-r--r-- | subagents/models.lua | 233 | ||||
| -rw-r--r-- | subagents/paths.lua | 169 | ||||
| -rw-r--r-- | subagents/profiles.lua | 122 | ||||
| -rw-r--r-- | subagents/progress.lua | 260 | ||||
| -rw-r--r-- | subagents/run.lua | 80 | ||||
| -rw-r--r-- | subagents/spawn.lua | 523 | ||||
| -rw-r--r-- | subagents/toml_workflows.lua | 570 | ||||
| -rw-r--r-- | subagents/workflow.lua | 614 |
11 files changed, 3287 insertions, 0 deletions
diff --git a/subagents/frontmatter.lua b/subagents/frontmatter.lua new file mode 100644 index 0000000..94fe2ac --- /dev/null +++ b/subagents/frontmatter.lua @@ -0,0 +1,96 @@ +-- Split a Markdown profile into its YAML frontmatter and its body. +-- +-- The format is the common one other agent harnesses use: if the very first +-- line of the file is exactly `---`, everything up to the next line that is +-- exactly `---` is a YAML mapping, and everything after that closing fence is +-- the body. Trailing carriage returns are tolerated so CRLF files parse. +-- +-- Anything unusual degrades to "no metadata, body only" with a warning rather +-- than an error, because the body is the part the user cannot afford to lose: +-- +-- * no opening fence -> the whole file is the body, no warning +-- * unterminated opening fence-> the whole file is the body, no warning +-- (a lone `---` at the top of a prose file is a horizontal rule, not a +-- broken header, so this case is deliberately silent) +-- * empty fenced block -> empty mapping, no warning +-- * lyaml missing or erroring -> body after the fence, warning returned +-- * YAML document not a map -> body after the fence, warning returned +-- +-- In the warning cases the fenced block is dropped rather than folded back +-- into the body: an unparseable header is noise the child agent should not be +-- asked to read. The body itself is never rewritten — no trimming, no +-- normalisation — so a prompt round-trips verbatim. + +local M = {} + +local function trim(s) + return (s:gsub("^%s+", ""):gsub("%s+$", "")) +end + +-- Iterate lines, yielding the line plus its start offset and the offset just +-- past its newline, so the caller can slice the original text exactly. +local function lines(text) + local pos = 1 + return function() + if pos > #text then + return nil + end + local start = pos + local nl = text:find("\n", pos, true) + local line + if nl then + line = text:sub(start, nl - 1) + pos = nl + 1 + else + line = text:sub(start) + pos = #text + 1 + end + return line, start, pos + end +end + +-- parse(text) -> data|nil, body, warning|nil +function M.parse(text) + if type(text) ~= "string" or text == "" then + return nil, "", nil + end + + local next_line = lines(text) + local first, _, after_first = next_line() + if first == nil or trim(first) ~= "---" then + return nil, text, nil + end + + local block_stop, body_start + for line, start, after in next_line do + if trim(line) == "---" then + block_stop = start - 1 + body_start = after + break + end + end + if body_start == nil then + return nil, text, nil + end + + local block = text:sub(after_first, block_stop) + local body = text:sub(body_start) + if trim(block) == "" then + return {}, body, nil + end + + local ok_lyaml, lyaml = pcall(require, "lyaml") + if not ok_lyaml then + return nil, body, "lyaml is not installed; ignoring the YAML frontmatter" + end + local ok, data = pcall(lyaml.load, block) + if not ok then + return nil, body, "YAML frontmatter did not parse: " .. tostring(data) + end + if type(data) ~= "table" then + return nil, body, "YAML frontmatter is not a mapping; ignoring it" + end + return data, body, nil +end + +return M diff --git a/subagents/jobs.lua b/subagents/jobs.lua new file mode 100644 index 0000000..a8cde6d --- /dev/null +++ b/subagents/jobs.lua @@ -0,0 +1,427 @@ +-- Start child agent jobs, bound how many run at once, and drain their events. +-- +-- One module owns the session-wide concurrency bound and the event pump, so +-- the policy that decides *what* to start (subagents/spawn.lua) never touches +-- luv, and the callers that wait for a result (subagents/run.lua, the workflow +-- API) never touch a job. +-- +-- Threading and ownership: everything here runs on panto's Lua owner thread. +-- A job's pump runs on its own thread inside the binding; it buffers events on +-- the job and writes one byte to the wake pipe handed to it here. That byte is +-- 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. Where there is no coroutine +-- to park (a plain script) or a job with no wake pipe, the identical drain +-- runs in a loop instead — same gate, same settle bookkeeping, only the wait +-- differs. +-- +-- A settled result is read and cached the moment it appears, because +-- `job:close()` frees it. Jobs are otherwise left open until close_all() ends +-- the turn, so a result can still be read after its coroutine resumed. + +local ok_uv, uv = pcall(require, "luv") + +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 + +local handle_mt = {} +handle_mt.__index = handle_mt +handle_mt.__name = "subagents.job" + +-- Every handle that has not been closed, in start order; the queue is the +-- subset waiting for a slot; `running` counts the started-but-unsettled ones. +local live = {} +local queued = {} +local waiters = {} +local running = 0 +local pumping = false + +-- --------------------------------------------------------------------------- +-- Wake pipes +-- --------------------------------------------------------------------------- + +local function open_pipe() + if not ok_uv or type(uv.pipe) ~= "function" then + return nil + end + local ok, pair = pcall(uv.pipe, { nonblock = true }, { nonblock = true }) + if not ok or type(pair) ~= "table" then + return nil + end + return pair +end + +-- Close the poll before the fds: the pump has already exited by the time a job +-- settles, so nothing can be mid-write when the read end goes away. +local function close_pipe(handle) + if handle.poll then + pcall(handle.poll.stop, handle.poll) + pcall(handle.poll.close, handle.poll) + handle.poll = nil + end + if handle.fds then + pcall(uv.fs_close, handle.fds.read) + pcall(uv.fs_close, handle.fds.write) + handle.fds = nil + end +end + +local function drain_pipe(handle) + if not handle.fds then + return + end + while true do + local data = uv.fs_read(handle.fds.read, READ_CHUNK, -1) + if type(data) ~= "string" or #data < READ_CHUNK then + return + end + end +end + +-- --------------------------------------------------------------------------- +-- Gate, drain, settle +-- --------------------------------------------------------------------------- + +local drain +local pump_queue +local wake + +local function settle(handle, raw) + if handle.settled ~= nil then + return + end + local result = raw + if handle.spec.shape then + local ok, shaped = pcall(handle.spec.shape, raw) + if ok and type(shaped) == "table" then + result = shaped + elseif not ok then + result = { status = "failed", error = tostring(shaped), resumable = false } + end + end + handle.settled = result + if handle.state == "running" then + running = running - 1 + end + handle.state = "settled" + close_pipe(handle) + pump_queue() +end + +local function launch(handle) + handle.fds = open_pipe() + local ok, job, err = pcall(handle.spec.build, handle.fds and handle.fds.write or nil) + if not ok then + close_pipe(handle) + return false, tostring(job) + end + if not job then + close_pipe(handle) + return false, err and tostring(err) or "the child could not be started" + end + + handle.job = job + handle.state = "running" + running = running + 1 + + if handle.fds and ok_uv then + local armed, poll = pcall(uv.new_poll, handle.fds.read) + if armed and poll then + handle.poll = poll + poll:start("r", function() + drain(handle) + wake() + end) + end + end + return true +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 + return + end + pumping = true + while running < M.MAX_CONCURRENT do + local handle = table.remove(queued, 1) + if handle == nil then + break + end + if handle.state == "queued" then + local ok, err = launch(handle) + if not ok then + settle(handle, { status = "failed", error = tostring(err) }) + end + end + end + pumping = false +end + +-- Drain one job: the wake pipe, then every buffered event, then the result. +-- Returns true when anything moved, which is what the fallback wait loop uses +-- to decide whether it is spinning. +function drain(handle) + local job = handle.job + if job == nil or handle.settled ~= nil then + return false + end + drain_pipe(handle) + + local moved = false + local on_event = handle.spec.on_event + local event = job:next_event() + while event ~= nil do + moved = true + if on_event then + pcall(on_event, event) + end + event = job:next_event() + end + + local raw = job:result() + if raw ~= nil then + settle(handle, raw) + return true + end + return moved +end + +local function drain_all() + local moved = false + local index = 1 + while index <= #live do + if drain(live[index]) then + moved = true + end + index = index + 1 + end + return moved +end + +-- --------------------------------------------------------------------------- +-- Handles +-- --------------------------------------------------------------------------- + +function handle_mt:result() + return self.settled +end + +-- A queued child settles without its build ever running: no user message is +-- appended, so nothing on disk claims a turn that never happened. +function handle_mt:cancel() + if self.settled ~= nil then + return + end + if self.state == "queued" then + for index, other in ipairs(queued) do + if other == self then + table.remove(queued, index) + break + end + end + settle(self, { status = "cancelled", error = "cancelled before the child started" }) + return + end + if self.job then + pcall(self.job.request_cancel, self.job) + end +end + +-- start(startspec) -> handle | nil, err +-- +-- startspec = { +-- build = function(wake_fd) -> job | nil, err -- calls agent:run_async +-- label = string?, id = string?, one_shot = boolean? +-- on_event = function(event)? -- one call per drained run_async event +-- shape = function(raw) -> result? -- maps the settled run_async result +-- onto the caller's result table +-- } +-- +-- Over the bound the job is queued and `build` is not called yet. +function M.start(spec) + if type(spec) ~= "table" or type(spec.build) ~= "function" then + return nil, "jobs.start needs a build function" + end + + local handle = setmetatable({ + spec = spec, + label = spec.label, + id = spec.id, + one_shot = spec.one_shot == true, + state = "queued", + }, handle_mt) + live[#live + 1] = handle + + if running >= M.MAX_CONCURRENT then + queued[#queued + 1] = handle + return handle + end + + local ok, err = launch(handle) + if not ok then + for index, other in ipairs(live) do + if other == handle then + table.remove(live, index) + break + end + end + return nil, err + end + -- Deliberately not drained here: starting a child must not settle it, and + -- the first wake byte is already in the pipe by the time anyone waits. + return handle +end + +-- True while a child with this id has a turn in flight. +function M.active(id) + if id == nil then + return false + end + for _, handle in ipairs(live) do + if handle.id == id and handle.settled == nil then + return true + end + end + return false +end + +-- --------------------------------------------------------------------------- +-- Awaiting +-- --------------------------------------------------------------------------- + +-- "all" is satisfied only when every handle has settled; "first" as soon as one +-- has. Both answer in input order, and "first" hands the rest back by identity +-- so a caller can await them again. +local function collect(handles, mode) + if mode == "first" then + local results, remaining = {}, {} + for _, handle in ipairs(handles) do + if handle.settled ~= nil then + results[#results + 1] = handle.settled + else + remaining[#remaining + 1] = handle + end + end + if #results == 0 and #remaining > 0 then + return nil + end + return results, remaining + end + + local results = {} + for index, handle in ipairs(handles) do + if handle.settled == nil then + return nil + end + results[index] = handle.settled + end + return results, {} +end + +-- Resume every parked await whose condition now holds. Never resumes the +-- running coroutine: a coroutine that is executing cannot also be parked. +function wake() + local self_co = coroutine.running() + local index = 1 + while index <= #waiters do + local waiter = waiters[index] + local results, remaining = collect(waiter.handles, waiter.mode) + if results ~= nil and waiter.co ~= self_co and coroutine.status(waiter.co) == "suspended" then + table.remove(waiters, index) + coroutine.resume(waiter.co, results, remaining) + else + index = index + 1 + end + end +end + +-- Parking is only safe when something will wake us: a started job with no wake +-- pipe is drained by asking it directly instead. +local function can_park(handles) + if not coroutine.isyieldable() then + return false + end + for _, handle in ipairs(handles) do + if handle.state == "running" and handle.poll == nil then + return false + end + end + return true +end + +-- await(handles, mode) -> results, remaining +function M.await(handles, mode) + mode = mode or "all" + if type(handles) ~= "table" then + error("jobs.await expects an array of job handles", 2) + end + if mode ~= "all" and mode ~= "first" then + error("jobs.await mode must be \"all\" or \"first\"", 2) + end + + while true do + local moved = drain_all() + local results, remaining = collect(handles, mode) + if results ~= nil then + return results, remaining + end + wake() + + if can_park(handles) then + waiters[#waiters + 1] = { co = coroutine.running(), handles = handles, mode = mode } + local woken, rest = coroutine.yield() + if woken ~= nil then + return woken, rest + end + elseif not moved and ok_uv then + -- Nothing to drain and nothing to park on: yield the core rather + -- than burn it while the pump threads work. + uv.sleep(1) + end + end +end + +-- --------------------------------------------------------------------------- +-- Turn lifecycle +-- --------------------------------------------------------------------------- + +-- 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() + for index = #live, 1, -1 do + live[index]:cancel() + end +end + +-- The turn is over: join every pump and drop the state. Requests go out first +-- so the joins overlap instead of running one child's teardown at a time. +function M.close_all() + for _, handle in ipairs(live) do + if handle.job and handle.settled == nil then + pcall(handle.job.request_cancel, handle.job) + end + end + for _, handle in ipairs(live) do + if handle.job then + pcall(handle.job.close, handle.job) + handle.job = nil + end + close_pipe(handle) + handle.state = "closed" + end + live, queued, waiters = {}, {}, {} + running = 0 +end + +return M diff --git a/subagents/luatool.lua b/subagents/luatool.lua new file mode 100644 index 0000000..089c67b --- /dev/null +++ b/subagents/luatool.lua @@ -0,0 +1,193 @@ +-- 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. +-- +-- 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 +-- standard library plus `subagents.workflow`; it has no `os`, `io`, `debug`, +-- `package`, `require`, `load`, `dofile`, `coroutine`, `setmetatable`, or +-- `getmetatable`, and `print` is a no-op so generated code cannot scribble on +-- the TUI. `string`, `table`, `math`, and `utf8` are shallow copies, so a guest +-- that reassigns `table.insert` only breaks itself, and `string.dump` is +-- removed from the copy. +-- +-- `pcall`/`xpcall` are deliberately absent: they would let a guest catch the +-- instruction-budget error and spin again in fresh 10M-instruction chunks. A +-- guest has no need to recover from its own errors — the handler reports them — +-- and with no `pcall`, `coroutine`, or metatable access left, nothing in the +-- environment can trap the hook error before it reaches the host. +-- +-- Known, accepted gaps in that sandbox: +-- +-- * The real string metatable is still reachable through any string literal +-- (`("").dump`), so `string.dump` is obtainable. Without `load` there is no +-- way to turn bytecode back into a running function, so this is noise rather +-- than an escape. +-- * `ctx:agent` returns handles the guest can read and scribble on. Nothing +-- reachable from one starts work: the running job — which owns the child's +-- agent and could start turns outside the cap — lives in a private side table +-- in workflow.lua, as do the job cap, the counter, and the profile set, so a +-- guest holding `ctx` and its handles cannot raise its own cap or reach the +-- host. +-- +-- 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. + +local workflow = require("subagents.workflow") +local run = require("subagents.run") + +local MAX_JOBS = 32 +local INSTRUCTION_BUDGET = 10000000 +local CHUNK_NAME = "subagents.lua" + +local M = {} + +M.max_jobs = MAX_JOBS +M.instruction_budget = INSTRUCTION_BUDGET + +local function shallow_copy(source, skip) + local copy = {} + for key, value in pairs(source) do + if key ~= skip then + copy[key] = value + end + end + return copy +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 env = { + assert = assert, + error = error, + ipairs = ipairs, + next = next, + pairs = pairs, + select = select, + tonumber = tonumber, + tostring = tostring, + type = type, + string = shallow_copy(string, "dump"), + table = shallow_copy(table), + math = shallow_copy(math), + utf8 = shallow_copy(utf8), + print = function() end, + subagents = { workflow = workflow.workflow }, + } + env._G = env + return env +end + +M.build_env = build_env + +local function budget_hook() + error("instruction budget exceeded", 0) +end + +-- subagents.run's block formatter expects string output; a structured worker's +-- output is a decoded table, so it is re-encoded first. +local function format_one(result) + if type(result.output) == "table" then + local flattened = {} + for key, value in pairs(result) do + flattened[key] = value + end + flattened.output = workflow.output_text(result) + return run.format_result(flattened) + end + return run.format_result(result) +end + +-- Format whatever the workflow callback returned. Result-shaped tables (the +-- common case: one settled result, or an array of them) render as the same +-- plain "key: value" block subagents.run uses; anything else is encoded +-- compactly so the model still sees it. +local function format_return(value) + if value == nil then + return "The workflow returned no value." + end + if type(value) ~= "table" then + return tostring(value) + end + if value.status ~= nil then + return format_one(value) + end + local blocks, count = {}, 0 + for index, entry in ipairs(value) do + if type(entry) ~= "table" or entry.status == nil then + blocks = nil + break + end + blocks[index] = format_one(entry) + count = count + 1 + end + if blocks and count > 0 then + return table.concat(blocks, "\n\n") + end + return workflow.json_encode(value) +end + +M.format_return = format_return + +-- 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) + 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 + + local chunk, load_err = load(input.source, CHUNK_NAME, "t", build_env()) + if not chunk then + return "Error: source did not compile: " .. tostring(load_err) + end + + local built_ok, built = pcall(chunk) + 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, + 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) +end + +return M diff --git a/subagents/models.lua b/subagents/models.lua new file mode 100644 index 0000000..511aa51 --- /dev/null +++ b/subagents/models.lua @@ -0,0 +1,233 @@ +-- 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: +-- +-- { 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 +-- the whole registry into the conversation. +-- +-- The host owns the catalog and the provider-specific reasoning rules, +-- including effort levels a Lua protocol reports dynamically. Anything this +-- 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 + +local M = {} + +local function host() + return require("panto").ext +end + +local function ask(q) + local ok, result = pcall(host().models, q) + if not ok then + return nil, tostring(result) + end + if type(result) ~= "table" then + return nil, "the host returned no model information" + end + return result +end + +local function format_overview(result) + local out = { + "inherited model: " .. (result.model or "(unknown)"), + "inherited reasoning: " .. (result.reasoning or "(provider default)"), + } + local providers = result.providers or {} + if #providers == 0 then + out[#out + 1] = "providers: (none configured)" + return table.concat(out, "\n") + end + out[#out + 1] = "providers:" + for _, provider in ipairs(providers) do + out[#out + 1] = string.format(" %s (%s, %d models)", + tostring(provider.name), tostring(provider.style or "?"), tonumber(provider.models) or 0) + end + return table.concat(out, "\n") +end + +local function format_match(match) + local detail = {} + if match.wire_model then + detail[#detail + 1] = "wire " .. tostring(match.wire_model) + end + if match.reasoning_default then + detail[#detail + 1] = "default reasoning " .. tostring(match.reasoning_default) + end + if match.context_window then + detail[#detail + 1] = "context " .. tostring(match.context_window) + end + if match.max_tokens then + detail[#detail + 1] = "max tokens " .. tostring(match.max_tokens) + end + local line = " " .. tostring(match.ref) + if #detail > 0 then + line = line .. " — " .. table.concat(detail, ", ") + end + return line +end + +local function format_exact(result, ref) + if not result.found then + return string.format( + "No configured model matches '%s'. Search with subagents.models { query = \"...\" }.", ref) + end + local out = { "model: " .. tostring(result.ref or ref) } + if result.wire_model then + out[#out + 1] = "wire model: " .. tostring(result.wire_model) + end + out[#out + 1] = "default reasoning: " .. (result.reasoning_default or "(provider default)") + local levels = result.reasoning_levels + if type(levels) == "table" and #levels > 0 then + out[#out + 1] = "reasoning levels: " .. table.concat(levels, ", ") + end + if result.context_window then + out[#out + 1] = "context window: " .. tostring(result.context_window) + end + if result.max_tokens then + out[#out + 1] = "max tokens: " .. tostring(result.max_tokens) + end + return table.concat(out, "\n") +end + +local function format_search(result) + local matches = result.matches or {} + if #matches == 0 then + return "No models match that query." + end + local out = { string.format("%d match(es):", #matches) } + for _, match in ipairs(matches) do + out[#out + 1] = format_match(match) + end + if result.truncated then + out[#out + 1] = string.format("%d shown, more exist — refine the query", #matches) + end + return table.concat(out, "\n") +end + +local function optional_string(value, field) + if value == nil then + return nil, nil + end + if type(value) ~= "string" or value == "" then + return nil, string.format("Error: `%s` must be a non-empty string when given.", field) + end + return value, nil +end + +local function clamp_limit(value) + local limit = tonumber(value) or DEFAULT_LIMIT + limit = math.floor(limit) + if limit < 1 then + return 1 + end + if limit > MAX_LIMIT then + return MAX_LIMIT + end + 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) + 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") + if err then + return err + end + local provider + provider, err = optional_string(input.provider, "provider") + if err then + return err + end + local text + text, err = optional_string(input.query, "query") + if err then + 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 + return "Error: " .. ask_err + end + return format_exact(result, model) + end + + if provider or text then + local result, ask_err = ask({ provider = provider, query = text, limit = clamp_limit(input.limit) }) + if not result then + return "Error: " .. ask_err + end + return format_search(result) + end + + local result, ask_err = ask({}) + if not result then + return "Error: " .. ask_err + end + return format_overview(result) +end + +return M diff --git a/subagents/paths.lua b/subagents/paths.lua new file mode 100644 index 0000000..678e241 --- /dev/null +++ b/subagents/paths.lua @@ -0,0 +1,169 @@ +-- Filesystem and session-path helpers shared by the subagents extension. +-- +-- Two jobs live here: +-- +-- 1. Config-layer discovery. `config_roots("agents")` returns the two +-- 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. +-- +-- 2. The child session store. `child_store_dir()` derives +-- `<session_dir>/subagents/<primary session id>` from the host's +-- `session_info()`, and `ensure_dir` creates it before a store is opened +-- there. Nothing here duplicates panto's XDG / PANTO_SESSION_DIR / +-- cwd-encoding logic; the per-cwd session directory arrives ready-made +-- from the host. +-- +-- Edge cases: a missing or unreadable directory yields no files rather than an +-- error, because both config layers are optional. Directory recursion is +-- capped (MAX_DEPTH) so a symlink cycle cannot hang discovery. Entries whose +-- type the platform does not report during scandir are stat'ed individually, +-- which also means a symlinked directory is followed like a real one. Results +-- are sorted so discovery order — and therefore shadowing — is deterministic. +-- +-- `luv` is resolved with pcall: only the filesystem helpers need it, so a +-- host or test run without luv can still use the rest of the extension and +-- gets a clear error if it actually walks the filesystem. + +local ok_uv, uv = pcall(require, "luv") + +local MAX_DEPTH = 16 + +local M = {} + +local function host() + return require("panto").ext +end + +local function require_uv() + if not ok_uv then + error("panto-subagents: the 'luv' module is required for filesystem discovery", 2) + end + return uv +end + +-- Current working directory, i.e. the project root of this panto session. +function M.cwd() + return require_uv().cwd() +end + +-- Extension-less basename of a path: "/a/b/reviewer.md" -> "reviewer". +function M.stem(path) + local base = path:match("[^/]+$") or path + return (base:gsub("%.[^.]+$", "")) +end + +-- Whole-file read. Returns nil, err for a file that cannot be opened. +function M.read_file(path) + local fh, err = io.open(path, "r") + if not fh then + return nil, err or ("could not open " .. path) + end + local data = fh:read("a") + fh:close() + if data == nil then + return nil, "could not read " .. path + end + return data +end + +-- User layer first, project layer second: later roots shadow earlier ones. +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 + end + local cwd = M.cwd() + if cwd ~= nil and cwd ~= "" then + roots[#roots + 1] = cwd .. "/.panto/" .. name + end + return roots +end + +-- Every file at or below `root` whose name ends with `suffix`, sorted. +function M.walk(root, suffix) + local lib = require_uv() + local found = {} + + local function visit(dir, depth) + if depth > MAX_DEPTH then + return + end + local req = lib.fs_scandir(dir) + if not req then + return + end + while true do + local name, kind = lib.fs_scandir_next(req) + if not name then + break + end + local path = dir .. "/" .. name + if kind ~= "directory" and kind ~= "file" then + local stat = lib.fs_stat(path) + kind = stat and stat.type or kind + end + if kind == "directory" then + visit(path, depth + 1) + elseif kind == "file" and name:sub(-#suffix) == suffix then + found[#found + 1] = path + end + end + end + + visit(root, 1) + table.sort(found) + return found +end + +-- Create `path` and every missing parent, tolerating one that already exists. +-- A store cannot be opened on a directory that is not there yet, and the child +-- catalog is two levels below a session directory panto may itself have only +-- just made. Returns nil, err rather than raising: this runs inside a spawn, +-- where a failure is the child's failure to report. +function M.ensure_dir(path) + if not ok_uv then + return nil, "panto-subagents: the 'luv' module is required to create the child store directory" + end + if type(path) ~= "string" or path == "" then + return nil, "no child store directory to create" + end + local made = path:sub(1, 1) == "/" and "" or "." + for segment in path:gmatch("[^/]+") do + made = made .. "/" .. segment + if uv.fs_stat(made) == nil then + local ok, err = uv.fs_mkdir(made, 493) -- 0755 + -- A racing writer is fine; only a directory that still is not + -- there afterwards is a failure. + if not ok and uv.fs_stat(made) == nil then + return nil, tostring(err) + end + end + end + return true +end + +-- The per-primary child catalog: <session_dir>/subagents/<primary id>. +-- Returns the directory plus the session info it came from, so a caller that +-- also needs the owning session id does not ask twice. This is pure string +-- assembly; `ensure_dir` creates it at spawn time. +function M.child_store_dir() + local info = host().session_info() + if type(info) ~= "table" or type(info.session_dir) ~= "string" or type(info.session_id) ~= "string" then + return nil, "no primary session information available" + end + return info.session_dir .. "/subagents/" .. info.session_id, info +end + +return M diff --git a/subagents/profiles.lua b/subagents/profiles.lua new file mode 100644 index 0000000..4ae9f91 --- /dev/null +++ b/subagents/profiles.lua @@ -0,0 +1,122 @@ +-- Discover agent profiles: Markdown files with YAML frontmatter. +-- +-- Two layers are read, lowest precedence first: +-- +-- 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. +-- +-- Recognised frontmatter keys, all optional: +-- +-- name the profile's identity; defaults to the filename stem +-- description one line shown to the primary in the subagents.run schema +-- model full `provider:model` only +-- reasoning passed through as written; the runtime validates it +-- +-- Unknown keys are ignored so a profile written for another harness still +-- loads. A `model` that is not Panto's `provider:model` syntax is dropped with +-- a warning and the child inherits the primary model — a foreign model +-- spelling must not cost the user a working prompt. Nothing else here is +-- fatal either: an unreadable file, broken YAML, or a duplicate name inside +-- one layer produces a warning and discovery continues. Warnings are returned +-- rather than logged because an extension has no logging channel; the caller +-- decides where they surface. +-- +-- Files are read eagerly. Profiles are small and the whole set is needed to +-- build the tool description at activation anyway, so lazy bodies would buy +-- nothing. + +local frontmatter = require("subagents.frontmatter") +local paths = require("subagents.paths") + +local MODEL_PATTERN = "^[^%s:]+:[^%s:]+$" + +local M = {} + +local function trim(s) + return (s:gsub("^%s+", ""):gsub("%s+$", "")) +end + +local function string_field(data, key) + local value = data and data[key] + if type(value) ~= "string" then + return nil + end + value = trim(value) + if value == "" then + return nil + end + return value +end + +-- Build one profile from a file. Appends to `warnings` on anything odd. +local function load_profile(path, layer, warnings) + local text, err = paths.read_file(path) + if not text then + warnings[#warnings + 1] = string.format("profile %s: %s", path, tostring(err)) + return nil + end + + local data, body, warning = frontmatter.parse(text) + local name = string_field(data, "name") or paths.stem(path) + if warning then + warnings[#warnings + 1] = string.format("profile %s: %s", name, warning) + end + + local model = string_field(data, "model") + if model and not model:match(MODEL_PATTERN) then + warnings[#warnings + 1] = + string.format("profile %s: ignoring model '%s' (not provider:model)", name, model) + model = nil + end + + return { + name = name, + description = string_field(data, "description") or "", + model = model, + reasoning = string_field(data, "reasoning"), + body = body, + path = path, + layer = layer, + } +end + +-- discover(roots) -> { list = {...}, by_name = {...}, warnings = {...} } +-- +-- `roots` defaults to the two config layers and exists so tests can point +-- discovery at temporary directories. `list` is sorted by name. +function M.discover(roots) + roots = roots or paths.config_roots("agents") + + local by_name = {} + local warnings = {} + + for _, root in ipairs(roots) do + local seen = {} + for _, path in ipairs(paths.walk(root, ".md")) do + local profile = load_profile(path, root, warnings) + if profile then + if seen[profile.name] then + warnings[#warnings + 1] = string.format( + "profile %s: %s shadows %s in the same layer", + profile.name, path, seen[profile.name]) + end + seen[profile.name] = path + by_name[profile.name] = profile + end + end + end + + local list = {} + for _, profile in pairs(by_name) do + list[#list + 1] = profile + end + table.sort(list, function(a, b) return a.name < b.name end) + + return { list = list, by_name = by_name, warnings = warnings } +end + +return M diff --git a/subagents/progress.lua b/subagents/progress.lua new file mode 100644 index 0000000..274e2a8 --- /dev/null +++ b/subagents/progress.lua @@ -0,0 +1,260 @@ +-- Live progress cards for in-flight children. +-- +-- While the primary model is blocked on one of the subagents.* tools, its +-- children are streaming. This folds each child's event stream into one small +-- card and renders the set as the component of the tool-call entry that +-- started them, so several concurrent children stay visually distinct and the +-- cards disappear with the entry they belong to. The events are presentation +-- data only: nothing here ever reaches a conversation. +-- +-- Linking a card to the right entry. The host fires `tool_call_complete` with +-- the tool-use id just before dispatching that call, and hands the same id to +-- the handler as `context.tool_call_id`; `claim` records the board under the +-- id, `bind` ties the running handler coroutine to it, and `card` finds the +-- board of whichever coroutine is asking. A workflow callback runs on its +-- handler's own coroutine, so a card raised deep inside one still lands on the +-- entry the model can see. +-- +-- Everything degrades to a no-op card: in print mode, in a test, or under a +-- host with no component machinery there is no board to attach to, and a child +-- still runs exactly the same. +-- +-- Rendering is deliberately small — a four-line ring per card, truncated to the +-- terminal width — because several children run at once and the panel must not +-- push the transcript off the screen. + +local MAX_CARD_LINES = 4 +local MAX_LINE_BYTES = 1024 +local BODY_INDENT = " " +local TOOL_PREFIX = "subagents." + +local GLYPHS = { + running = "◷", + completed = "✔", + failed = "✖", + cancelled = "⊘", +} + +local M = {} + +-- Boards keyed by tool-call id, and the board bound to each handler coroutine. +-- Weak keys on the second so a finished handler's binding disappears with it. +local boards = {} +local bound = setmetatable({}, { __mode = "k" }) + +-- --------------------------------------------------------------------------- +-- Text handling +-- --------------------------------------------------------------------------- + +-- Control bytes would move the cursor or confuse the renderer's width +-- accounting, so they become spaces. UTF-8 continuation bytes are >= 0x80 and +-- pass through untouched. +local function sanitize(text) + return (text:gsub("[%z\1-\31\127]", " ")) +end + +local function truncate(text, width) + if width == nil or width < 1 then + return text + end + local length = utf8.len(text) + if length == nil or length <= width then + return text + end + return text:sub(1, (utf8.offset(text, width + 1) or (#text + 1)) - 1) +end + +-- --------------------------------------------------------------------------- +-- Cards +-- --------------------------------------------------------------------------- + +local card_mt = {} +card_mt.__index = card_mt + +local function repaint(card) + local board = card.board + if board and board.handle then + pcall(board.handle.invalidate, board.handle) + end +end + +local function adopt(card, line) + card.lines[#card.lines + 1] = line + while #card.lines > MAX_CARD_LINES do + table.remove(card.lines, 1) + end +end + +-- A discrete, already-complete line: a tool marker, a model label, an error. +local function marker(card, prefix, text) + card.partial = false + if text == nil or text == "" then + return + end + adopt(card, prefix .. sanitize(text:sub(1, MAX_LINE_BYTES))) +end + +-- Fold a text delta in, breaking it on newlines. A chunk with no newline leaves +-- the tail line open so the next delta continues it. +local function append_text(card, text) + local rest = text + while rest ~= "" do + local newline = rest:find("\n", 1, true) + local chunk = newline and rest:sub(1, newline - 1) or rest + if card.partial and #card.lines > 0 then + local index = #card.lines + local grown = card.lines[index] .. sanitize(chunk) + card.lines[index] = grown:sub(1, MAX_LINE_BYTES) + elseif chunk ~= "" then + adopt(card, sanitize(chunk:sub(1, MAX_LINE_BYTES))) + end + card.partial = newline == nil + rest = newline and rest:sub(newline + 1) or "" + end +end + +-- One run_async event. `content_delta` carries only an index, so the block type +-- comes from the `block_start` that opened it — the same flag the v0 pump kept. +function card_mt:event(event) + if type(event) ~= "table" then + return + end + local kind = event.type + if kind == "block_start" then + self.text_block = event.block_type == "text" + elseif kind == "content_delta" then + if self.text_block and type(event.delta) == "string" then + append_text(self, event.delta) + end + elseif kind == "tool_details" then + marker(self, "⚒ ", event.name) + elseif kind == "tool_dispatch_complete" or kind == "message_complete" then + self.partial = false + else + return + end + repaint(self) +end + +-- The child settled. Only a failure or a cancellation gets a closing line; a +-- completed child's report is the tool result the model already sees. +function card_mt:done(status, message) + self.status = GLYPHS[status] and status or "failed" + if self.status ~= "completed" then + marker(self, "", message) + else + self.partial = false + end + repaint(self) +end + +local NOOP = setmetatable({ lines = {} }, { + __index = { + event = function() end, + done = function() end, + }, +}) + +-- --------------------------------------------------------------------------- +-- Boards (one per tool-call entry) +-- --------------------------------------------------------------------------- + +local function render(board, width) + local out = {} + if #board.cards == 0 or (width or 0) <= 4 then + return out + end + out[#out + 1] = "" + for _, card in ipairs(board.cards) do + local sid = card.sid ~= "" and (" " .. card.sid) or "" + out[#out + 1] = truncate(string.format("%s %s%s — %s", + GLYPHS[card.status] or GLYPHS.running, card.label, sid, card.status), width) + for _, line in ipairs(card.lines) do + out[#out + 1] = BODY_INDENT .. truncate(line, width - #BODY_INDENT) + end + end + return out +end + +local function new_board() + local board = { cards = {}, handle = nil } + board.component = { + render = function(_, width) + return render(board, width) + end, + } + return board +end + +-- --------------------------------------------------------------------------- +-- Host wiring +-- --------------------------------------------------------------------------- + +-- `tool_call_complete` for one of our tools: claim that entry's component so +-- the cards the handler is about to raise have somewhere to render. +function M.claim(event) + -- The event is host userdata; a host that predates any of these fields + -- answers nil, and one that predates the whole object cannot be indexed. + local ok, name = pcall(function() + return event.tool_name + end) + if not ok or type(name) ~= "string" or name:sub(1, #TOOL_PREFIX) ~= TOOL_PREFIX then + return + end + if type(event.set_component) ~= "function" then + return + end + + local key = event.id or name + local board = new_board() + local attached, handle = pcall(event.set_component, event, board.component) + if not attached then + return + end + board.handle = handle + boards[key] = board +end + +-- Tie the running handler coroutine to the entry it was dispatched for. +function M.bind(context) + local key = type(context) == "table" and context.tool_call_id or nil + if key == nil then + return + end + bound[coroutine.running()] = key +end + +-- card(label, id, detail) -> card +-- +-- The card of the entry whose handler is running, or an inert one when there is +-- no such entry (print mode, a plain script, a host without components). +function M.card(label, id, detail) + local key = bound[coroutine.running()] + local board = key and boards[key] + if board == nil or (board.handle and board.handle.alive and not board.handle:alive()) then + return NOOP + end + + local card = setmetatable({ + board = board, + label = tostring(label or "subagent"), + sid = type(id) == "string" and id:sub(1, 8) or "", + status = "running", + lines = {}, + partial = false, + text_block = false, + }, card_mt) + board.cards[#board.cards + 1] = card + marker(card, "↳ ", detail) + repaint(card) + return card +end + +-- End of turn: the entries are gone, and the primary's own tool results are the +-- durable record from here on. +function M.reset() + boards = {} + bound = setmetatable({}, { __mode = "k" }) +end + +return M diff --git a/subagents/run.lua b/subagents/run.lua new file mode 100644 index 0000000..375af91 --- /dev/null +++ b/subagents/run.lua @@ -0,0 +1,80 @@ +-- 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. +-- +-- The call blocks until the child settles. Parallelism is ordinary tool +-- batching: several subagents.run calls emitted in one batch run concurrently +-- under the session-wide bound in subagents/jobs.lua, and one failure does not +-- disturb its siblings. +-- +-- Failures are values, not exceptions. Everything the model could plausibly +-- have caused — a missing prompt, both selectors at once, an unknown profile, +-- an unresumable id, a child that errored or was cancelled — comes back as +-- readable text. Validation that fails before a child is allocated has +-- no id to report, and a new child that dies before its first assistant +-- message has no durable file, so it reports `resumable: false` rather than +-- promising a continuation that would not resolve. +-- +-- The result block is plain `key: value` lines rather than JSON: it is read by +-- a model, and the field names match the design's result shape (id, agent, +-- status, resumable, then the output or the error message). + +local jobs = require("subagents.jobs") +local spawn = require("subagents.spawn") + +local M = {} + +-- A resumed child has no profile in hand; its identity comes back from the +-- manifest metadata stored on its first profile system message. +local function manifest_agent(result) + local manifest = result.manifest + if type(manifest) ~= "table" then + return nil + end + local mine = manifest.subagents + if type(mine) ~= "table" or type(mine.agent) ~= "string" then + return nil + end + return mine.agent +end + +-- format_result(result, agent_name) -> the model-visible block. +function M.format_result(result, agent_name) + local body = result.output + if body == nil or body == "" then + body = result.error or "" + end + return table.concat({ + "id: " .. (result.id or "(none)"), + "agent: " .. (agent_name or manifest_agent(result) or "?"), + "status: " .. (result.status or "unknown"), + "resumable: " .. tostring(result.resumable == true), + "--- output ---", + tostring(body), + }, "\n") +end + +function M.handle(input, profiles) + local spec, err = spawn.build_spec(input, profiles) + if not spec then + return "Error: " .. tostring(err) + end + + local handle, spawn_err = spawn.spawn(spec) + if not handle then + return "Error: " .. tostring(spawn_err) + end + + local results = jobs.await({ handle }, "all") + local result = results and results[1] + if type(result) ~= "table" then + return "Error: the subagent produced no result." + end + + return M.format_result(result, spec.label) +end + +return M diff --git a/subagents/spawn.lua b/subagents/spawn.lua new file mode 100644 index 0000000..a83e442 --- /dev/null +++ b/subagents/spawn.lua @@ -0,0 +1,523 @@ +-- Turn a delegation request into a child agent, and start its turn. +-- +-- Every path that starts a child — the subagents.run tool, `ctx:agent` in the +-- Lua workflow API, and the TOML workflow lowering — goes through here, so the +-- validation rules and the model/reasoning precedence exist exactly once: +-- +-- model = call.model or profile.model or (inherited) +-- reasoning = call.reasoning or profile.reasoning or (inherited) +-- +-- "Inherited" means the field is absent from the spec, and the primary's live +-- values from `session_info()` apply. A resumed child reads its own last +-- effective values from the stored conversation instead of a profile, so a +-- continuation without overrides keeps running on what it ran on before. The +-- 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), which is how a resumed child re-identifies itself. The parent +-- dialogue is never copied, which is why the child-role text tells the child +-- its final message is the whole of what the delegator sees. +-- +-- Edge cases: a profile with an empty body contributes no system message and +-- therefore no manifest, so a resumed child started from a body-less profile +-- reports no agent name. Resume opens the stored conversation as canonical, so +-- profile edits never reach an existing child. Errors are returned as plain +-- lowercase messages without an "Error: " prefix; the tool layer decides how to +-- present them. Nothing here raises: the binding reports its failures by +-- raising, and every such call goes through `try` so a caller sees one shape. + +local jobs = require("subagents.jobs") +local paths = require("subagents.paths") +local profiles_mod = require("subagents.profiles") +local progress = require("subagents.progress") + +-- A child never gets the delegation tools themselves: no recursion. +local TOOL_PREFIX = "subagents." + +local M = {} + +M.CHILD_ROLE = table.concat({ + "You are a subagent working inside another agent's session.", + "Complete the task you are given directly and end with a clear,", + "self-contained report; your final message is returned to the", + "delegating agent verbatim. You cannot ask the user questions.", +}, " ") + +local discovered = nil + +local function host() + return require("panto").ext +end + +local function binding() + return require("panto") +end + +-- The binding reports every failure by raising. Route those through one place +-- so a host error becomes the `nil, message` shape the callers already handle. +local function try(fn, ...) + local ok, value = pcall(fn, ...) + if not ok then + return false, tostring(value) + end + return true, value +end + +local function nonempty(value) + if type(value) == "string" and value ~= "" then + return value + end + return nil +end + +-- Discovery is cached: activation discovers once, and callers that omit the +-- profile set (a workflow calling build_spec with one argument) reuse it. +function M.profiles(given) + if given ~= nil then + return given + end + if discovered == nil then + discovered = profiles_mod.discover() + end + return discovered +end + +-- Comma-joined sorted profile names, for "unknown agent" messages. +function M.agent_names(profiles) + profiles = M.profiles(profiles) + local names = {} + for name in pairs(profiles.by_name or {}) do + names[#names + 1] = name + end + if #names == 0 then + return "(no agent profiles found)" + end + table.sort(names) + return table.concat(names, ", ") +end + +local function optional_string(value, field) + if value == nil then + return nil, nil + end + if type(value) ~= "string" or value == "" then + return nil, string.format("`%s` must be a non-empty string when given", field) + end + return value, nil +end + +local function build_output(output) + if type(output) ~= "table" then + return nil, "`output` must be a table" + end + if type(output.schema) ~= "table" then + return nil, "`output.schema` must be a JSON-Schema table" + end + return { + name = output.name or "emit_result", + description = output.description, + schema = output.schema, + }, nil +end + +-- build_spec(input, profiles) -> spec | nil, err +-- +-- input = { agent | id, prompt, model?, reasoning?, output? } +function M.build_spec(input, profiles) + if type(input) ~= "table" then + return nil, "expected a table of arguments" + end + if type(input.prompt) ~= "string" or input.prompt:match("^%s*$") then + return nil, "`prompt` must be a non-empty string" + end + + local agent, err = optional_string(input.agent, "agent") + if err then + return nil, err + end + local id + id, err = optional_string(input.id, "id") + 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" + end + if not agent and not id then + return nil, "pass exactly one of `agent` (start a new child) or `id` (continue one)" + end + + local model + model, err = optional_string(input.model, "model") + if err then + return nil, err + end + local reasoning + reasoning, err = optional_string(input.reasoning, "reasoning") + if err then + return nil, err + end + + local output + if input.output ~= nil then + output, err = build_output(input.output) + if err then + return nil, err + end + end + + -- The profile is resolved before the store is opened so an unknown agent + -- name reports itself instead of a session-directory failure. + local profile + if agent then + profile = (M.profiles(profiles).by_name or {})[agent] + if not profile then + return nil, string.format("unknown agent '%s'; known: %s", agent, M.agent_names(profiles)) + end + end + + -- child_store_dir returns the session info alongside the directory on + -- success, and the failure message in that same slot on failure. + local store_dir, info_or_err = paths.child_store_dir() + if not store_dir then + return nil, tostring(info_or_err) + end + + local spec = { + store_dir = store_dir, + prompt = input.prompt, + model = model, + reasoning = reasoning, + output = output, + } + + if id then + spec.session_id = id + return spec + end + + spec.label = profile.name + spec.model = model or profile.model + spec.reasoning = reasoning or profile.reasoning + + local system_messages = { { text = M.CHILD_ROLE } } + if profile.body and profile.body:match("%S") then + system_messages[#system_messages + 1] = { + text = profile.body, + metadata = { subagents = { owner = info_or_err.session_id, agent = profile.name } }, + } + end + spec.system_messages = system_messages + + return spec +end + +-- The stored conversation is the only record a resumed child has of itself: the +-- first system message carrying metadata holds the manifest, and the last user +-- message whose metadata names this extension holds the model and reasoning its +-- previous turn resolved to. A message whose metadata is malformed is skipped, +-- not treated as an error. +local function read_stored(conv) + local ok, messages = try(conv.messages, conv) + if not ok or type(messages) ~= "table" then + return {}, nil + end + + local manifest + for index = 1, #messages do + if messages[index].role == "system" then + local metadata = conv:message_metadata(index) + if type(metadata) == "table" then + manifest = metadata + break + end + end + end + + local defaults = {} + for index = #messages, 1, -1 do + if messages[index].role == "user" then + local metadata = conv:message_metadata(index) + local mine = type(metadata) == "table" and metadata.subagents or nil + if type(mine) == "table" then + defaults.model = nonempty(mine.model) + defaults.reasoning = nonempty(mine.reasoning) + break + end + end + end + 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. +local function inherited_tools() + local primary = host().agent + if primary == nil then + return {} + end + local ok, decls = try(primary.tools, primary) + if not ok or type(decls) ~= "table" then + return {} + end + local kept = {} + for _, decl in ipairs(decls) do + if type(decl.name) ~= "string" or decl.name:sub(1, #TOOL_PREFIX) ~= TOOL_PREFIX then + kept[#kept + 1] = decl + end + end + return kept +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 }) + else + conv:add_system_message(message.text) + end + end + return true +end + +-- spawn(spec) -> handle | nil, err +-- +-- Resolve the model, open the child's store, build the agent, seed or reopen +-- its conversation, hand it the primary's tools, and start one turn under the +-- session-wide bound. Everything that can fail before the turn starts (an +-- unknown id, an id already busy, an unknown model, a store that cannot be +-- opened) fails here, so a caller either has a running child or a message. +function M.spawn(spec) + if type(spec) ~= "table" then + return nil, "expected a spawn spec" + end + local one_shot = type(spec.output) == "table" + if one_shot and spec.session_id then + return nil, "a structured-output child cannot be resumed" + end + if spec.session_id and jobs.active(spec.session_id) then + return nil, string.format("subagent '%s' already has a turn in flight", spec.session_id) + end + + local ext = host() + local panto = binding() + local _, info = try(ext.session_info) + info = type(info) == "table" and info or {} + + -- A structured worker is ephemeral by contract: no durable file to resume, + -- so it never touches the child catalog. + local opened, store + if one_shot then + opened, store = try(panto.null_store) + else + local made, dir_err = paths.ensure_dir(spec.store_dir) + if not made then + return nil, dir_err + end + opened, store = try(panto.file_system_jsonl_store, { dir = spec.store_dir }) + end + if not opened then + return nil, tostring(store) + end + if store == nil then + return nil, "the child session store could not be opened" + end + + local conv, defaults, manifest + if spec.session_id then + -- The ownership boundary is the primary's own catalog directory: an id + -- from another session simply is not in this store. + local unknown = string.format("unknown subagent id '%s' for this session", spec.session_id) + local asked, found = try(store.resolve, store, spec.session_id) + if not asked then + return nil, tostring(found) + end + if found == nil then + return nil, unknown + end + local loaded + loaded, conv = try(store.load, store, spec.session_id) + if not loaded then + return nil, tostring(conv) + end + if conv == nil then + return nil, unknown + end + defaults, manifest = read_stored(conv) + end + defaults = defaults or {} + + local model = spec.model or defaults.model or nonempty(info.model) + local reasoning = spec.reasoning or defaults.reasoning or nonempty(info.reasoning) + + local asked, cfg, resolve_err = pcall(ext.resolve_model, { + model = model, + reasoning = reasoning, + tool_choice = one_shot and { name = spec.output.name } or nil, + }) + if not asked then + return nil, tostring(cfg) + end + if cfg == nil then + return nil, resolve_err and tostring(resolve_err) or "the child model could not be resolved" + end + -- The labels the host actually resolved to are what the turn records, so a + -- continuation reads back the same spelling. + local model_label = nonempty(cfg.model) or model + local reasoning_label = nonempty(cfg.reasoning) or reasoning + + local built, agent = try(panto.agent, { + config = cfg, + store = store, + session_id = spec.session_id, + conversation = conv, + }) + if not built or agent == nil then + return nil, built and "the subagent could not be created" or tostring(agent) + end + + if not spec.session_id then + local seeded, seed_err = try(seed_conversation, agent, spec) + if not seeded then + return nil, seed_err + end + end + + local decls = inherited_tools() + if one_shot then + decls = { { + name = spec.output.name, + description = spec.output.description or "", + schema = spec.output.schema, + } } + end + local armed, tools_err = try(agent.set_tools, agent, decls) + if not armed then + return nil, tools_err + end + + -- A one-shot worker reports no id: it has no durable session to name. + local named, session_id = try(agent.session_id, agent) + local id = (not one_shot) and named and nonempty(session_id) or nil + + local card = progress.card(spec.label or "subagent", id, model_label) + + -- The settled run_async result becomes the result table every caller + -- already reads: run.lua's block and the workflow API's shape_result. + local function shape(raw) + raw = type(raw) == "table" and raw or {} + local result = { + id = id, + status = raw.status or "failed", + output = raw.text, + error = raw.error, + resumable = false, + model = model_label, + reasoning = reasoning_label, + manifest = manifest, + } + if one_shot then + local wanted = spec.output.name + for _, call in ipairs(raw.tool_calls or {}) do + if call.name == wanted then + result.structured_json = call.input + break + end + end + if result.structured_json == nil and result.status == "completed" then + result.status = "failed" + result.output = nil + result.error = string.format("the child did not call the required '%s' output tool", wanted) + end + elseif id then + -- Durable resumability is a fact about the store, not about the + -- status: a turn that died before its first assistant message left + -- nothing to continue from. Ask again now that it has settled. + local ok, found = try(store.resolve, store, id) + result.resumable = ok and found ~= nil + end + card:done(result.status, result.error) + return result + end + + local handle, start_err = jobs.start { + label = spec.label, + id = id, + one_shot = one_shot, + build = function(wake_fd) + local job, err = agent:run_async { + prompt = spec.prompt, + metadata = { subagents = { model = model_label, reasoning = reasoning_label } }, + dispatch_tools = not one_shot, + wake_fd = wake_fd, + } + if not job then + return nil, err or "the host could not start the subagent" + end + return job + end, + on_event = function(event) + card:event(event) + end, + shape = shape, + } + if not handle then + card:done("failed", start_err) + return nil, start_err + end + + -- The job borrows the agent, which borrows the store; anchor both on the + -- handle so neither is collected while the pump is running. + handle.agent = agent + handle.store = store + return handle +end + +return M diff --git a/subagents/toml_workflows.lua b/subagents/toml_workflows.lua new file mode 100644 index 0000000..fba1930 --- /dev/null +++ b/subagents/toml_workflows.lua @@ -0,0 +1,570 @@ +-- subagents/toml_workflows.lua +-- +-- Persistent TOML workflows: discovery, validation, execution, the generated +-- `/workflow:<name>` slash commands, and the model-facing `subagents.workflow` +-- tool. This is the fixed-DAG surface. Output-dependent branching and dynamic +-- 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). +-- +-- 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. +-- +-- 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. +-- +-- 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. + +local workflow = require("subagents.workflow") +local paths = require("subagents.paths") + +local M = {} + +-- toml2lua installs its module under the name "toml", not "toml2lua". +local TOML_MODULE = "toml" + +local function host() + return require("panto").ext +end + +local function load_toml() + local ok, toml = pcall(require, TOML_MODULE) + if not ok or type(toml) ~= "table" or type(toml.parse) ~= "function" then + return nil, "the 'toml2lua' rock is required to read TOML workflows" + end + return toml +end + +-- --------------------------------------------------------------------------- +-- Parsing and validation +-- --------------------------------------------------------------------------- + +local function is_array(value) + if type(value) ~= "table" then + return false + end + local count = 0 + for key in pairs(value) do + if type(key) ~= "number" then + return false + end + count = count + 1 + end + return count == #value +end + +local function optional_string(value, label) + if value == nil then + return nil, nil + end + if type(value) ~= "string" or value == "" then + return nil, label .. " must be a non-empty string when given" + end + return value, nil +end + +-- validate(def, fallback_name) -> normalized definition | nil, err +-- +-- 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 } }. +function M.validate(def, fallback_name) + if type(def) ~= "table" then + return nil, "workflow definition must be a table" + end + + local name, err = optional_string(def.name, "`name`") + if err then + return nil, err + end + name = name or fallback_name + if name == nil or name == "" then + return nil, "workflow has no name" + end + + local description + description, err = optional_string(def.description, "`description`") + if err then + return nil, err + end + + if not is_array(def.steps) or #def.steps == 0 then + return nil, "workflow '" .. name .. "' has no `steps` array" + end + + local steps, by_id = {}, {} + for index, raw in ipairs(def.steps) do + if type(raw) ~= "table" then + return nil, string.format("workflow '%s': step %d is not a table", name, index) + end + local where = string.format("workflow '%s' step %d", name, index) + if type(raw.id) ~= "string" or raw.id == "" then + return nil, where .. ": `id` is required and must be a non-empty string" + end + if by_id[raw.id] then + return nil, string.format("workflow '%s': duplicate step id '%s'", name, raw.id) + end + if type(raw.agent) ~= "string" or raw.agent == "" then + return nil, string.format("workflow '%s' step '%s': `agent` is required", name, raw.id) + end + if type(raw.prompt) ~= "string" or raw.prompt == "" then + return nil, string.format("workflow '%s' step '%s': `prompt` is required", name, raw.id) + end + + local model, model_err = optional_string(raw.model, "`model`") + if model_err then + return nil, string.format("workflow '%s' step '%s': %s", name, raw.id, model_err) + end + local reasoning, reasoning_err = optional_string(raw.reasoning, "`reasoning`") + if reasoning_err then + return nil, string.format("workflow '%s' step '%s': %s", name, raw.id, reasoning_err) + end + + local needs = {} + if raw.needs ~= nil then + if not is_array(raw.needs) then + return nil, string.format("workflow '%s' step '%s': `needs` must be an array", name, raw.id) + end + for _, need in ipairs(raw.needs) do + if type(need) ~= "string" or need == "" then + return nil, string.format("workflow '%s' step '%s': `needs` entries must be step ids", name, raw.id) + end + needs[#needs + 1] = need + end + end + + local step = { + id = raw.id, + agent = raw.agent, + prompt = raw.prompt, + model = model, + reasoning = reasoning, + needs = needs, + } + steps[#steps + 1] = step + by_id[raw.id] = step + end + + -- Dependencies must exist before the cycle walk, so an unknown name reports + -- itself rather than looking like a broken graph. + local terminal = {} + for _, step in ipairs(steps) do + terminal[step.id] = true + end + for _, step in ipairs(steps) do + for _, need in ipairs(step.needs) do + if not by_id[need] then + return nil, string.format("workflow '%s' step '%s': unknown dependency '%s'", name, step.id, need) + end + terminal[need] = nil + end + end + + -- Iterative-free DFS with a per-node mark: "open" means the node is on the + -- current path, so meeting it again is a cycle. + local mark = {} + local function visit(step, trail) + if mark[step.id] == "done" then + return true + end + if mark[step.id] == "open" then + return false, string.format( + "workflow '%s': dependency cycle through '%s' (%s)", + name, step.id, table.concat(trail, " -> ") .. " -> " .. step.id) + end + mark[step.id] = "open" + trail[#trail + 1] = step.id + for _, need in ipairs(step.needs) do + local ok, cycle_err = visit(by_id[need], trail) + if not ok then + return false, cycle_err + end + end + trail[#trail] = nil + mark[step.id] = "done" + return true + end + for _, step in ipairs(steps) do + local ok, cycle_err = visit(step, {}) + if not ok then + return nil, cycle_err + end + end + + return { + name = name, + description = description, + steps = steps, + by_id = by_id, + terminal = terminal, + } +end + +-- parse(text, fallback_name) -> definition | nil, err, declared_name +-- +-- On failure the third value is the `name` the document declared, when it read +-- as one, so discovery can index a broken file under the name it claims rather +-- than its filename stem — otherwise a broken project file would fail to shadow +-- the user workflow of the same name and the error would go unreported. +function M.parse(text, fallback_name) + local toml, err = load_toml() + if not toml then + return nil, err + end + local ok, parsed = pcall(toml.parse, text, { strict = true }) + if not ok then + return nil, "invalid TOML: " .. tostring(parsed) + end + if type(parsed) ~= "table" then + return nil, "invalid TOML: the document did not parse into a table" + end + local def, validate_err = M.validate(parsed, fallback_name) + if def then + return def + end + local declared = parsed.name + if type(declared) ~= "string" or declared == "" then + declared = nil + end + return nil, validate_err, declared +end + +-- --------------------------------------------------------------------------- +-- Discovery +-- --------------------------------------------------------------------------- + +-- discover() -> { list = ordered array, by_name = map, warnings = array } +-- +-- Later roots (the project layer) 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 +-- the name it declares (falling back to its stem only when it declares none), so +-- a broken project workflow reports its error rather than silently letting the +-- same-named user workflow run in its place. +function M.discover() + local list, by_name, warnings = {}, {}, {} + + -- Walking is the only part that can raise (a missing luv, a hostile + -- filesystem); a root that cannot be read contributes a warning and no + -- workflows, so discovery as a whole keeps its "never throws" contract. + local roots_ok, roots = pcall(paths.config_roots, "workflows") + if not roots_ok then + return { list = list, by_name = by_name, warnings = { tostring(roots) } } + end + + for _, root in ipairs(roots) do + local walk_ok, found = pcall(paths.walk, root, ".toml") + if not walk_ok then + warnings[#warnings + 1] = root .. ": " .. tostring(found) + found = {} + end + for _, path in ipairs(found) do + local stem = paths.stem(path) + local text, read_err = paths.read_file(path) + local entry + if not text then + entry = { name = stem, path = path, error = tostring(read_err) } + else + local def, err, declared = M.parse(text, stem) + if def then + entry = { name = def.name, path = path, definition = def } + else + entry = { name = declared or stem, path = path, error = tostring(err) } + end + end + if entry.error then + warnings[#warnings + 1] = path .. ": " .. entry.error + end + + local existing = by_name[entry.name] + if existing then + for index, candidate in ipairs(list) do + if candidate == existing then + list[index] = entry + break + end + end + else + list[#list + 1] = entry + end + by_name[entry.name] = entry + end + end + + return { list = list, by_name = by_name, warnings = warnings } +end + +-- --------------------------------------------------------------------------- +-- Lowering onto the Lua workflow API +-- --------------------------------------------------------------------------- + +local function dependency_text(result) + if result == nil then + return "[failed: not run]" + end + if result.status == "completed" then + return workflow.output_text(result) + end + 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. +local function step_prompt(step, input, settled) + local parts = { step.prompt, "\n\n## Workflow input\n\n", input } + for _, need in ipairs(step.needs) do + parts[#parts + 1] = "\n\n## Output of " .. need .. "\n\n" + parts[#parts + 1] = dependency_text(settled[need]) + end + return table.concat(parts) +end + +M.step_prompt = step_prompt + +-- lower(def) -> workflow object +function M.lower(def) + return workflow.workflow(function(ctx, input) + input = input or "" + local waiting = {} + for index, step in ipairs(def.steps) do + waiting[index] = step + end + + local settled = {} + local live, live_step = {}, {} + + -- One pass may unblock another (a skip cascades to its dependents), so + -- this repeats until nothing more can start or be skipped. + local function advance() + local changed = true + while changed do + changed = false + local index = 1 + while index <= #waiting do + local step = waiting[index] + local ready, skip = true, false + for _, need in ipairs(step.needs) do + local result = settled[need] + if result == nil then + ready = false + elseif result.status ~= "completed" then + skip = true + break + end + end + + if skip then + table.remove(waiting, index) + settled[step.id] = { + status = "skipped", + error = "skipped: a dependency did not complete", + } + changed = true + elseif ready then + table.remove(waiting, index) + local handle = ctx:agent({ + agent = step.agent, + prompt = step_prompt(step, input, settled), + model = step.model, + reasoning = step.reasoning, + }) + live[#live + 1] = handle + live_step[handle] = step.id + changed = true + else + index = index + 1 + end + end + end + end + + advance() + while #live > 0 do + local result, remaining = ctx:await(live, "first") + if result == nil then + break + end + local still = {} + for _, handle in ipairs(remaining or {}) do + still[handle] = true + end + for _, handle in ipairs(live) do + if not still[handle] then + settled[live_step[handle]] = result + break + end + end + live = remaining or {} + advance() + 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 + end + return out + end) +end + +-- run(def, input, profiles) -> array of terminal results +function M.run(def, input, profiles) + return workflow.execute(M.lower(def), input or "", { profiles = profiles }) +end + +-- --------------------------------------------------------------------------- +-- Model- and user-visible formatting +-- --------------------------------------------------------------------------- + +local function format_step(result) + return table.concat({ + "step: " .. tostring(result.id), + "status: " .. tostring(result.status), + "--- output ---", + workflow.output_text(result), + }, "\n") +end + +function M.format_results(results) + if type(results) ~= "table" or #results == 0 then + return "The workflow produced no terminal results." + end + local blocks = {} + for index, result in ipairs(results) do + blocks[index] = format_step(result) + end + return table.concat(blocks, "\n\n") +end + +-- --------------------------------------------------------------------------- +-- Tool and command entry points +-- --------------------------------------------------------------------------- + +local registry = nil + +-- The discovered set, discovered once per activation. +function M.workflows() + if registry == nil then + registry = M.discover() + end + return registry +end + +local function known_names(found) + local names = {} + for name in pairs(found.by_name) do + names[#names + 1] = name + end + if #names == 0 then + return "(no workflows found)" + end + table.sort(names) + return table.concat(names, ", ") +end + +local function run_named(name, input, profiles) + local found = M.workflows() + local entry = found.by_name[name] + if not entry then + return "Error: unknown workflow '" .. tostring(name) .. "'; known: " .. known_names(found) + end + if not entry.definition then + return "Error: workflow '" .. name .. "' failed to load: " .. tostring(entry.error) + end + local ok, results = pcall(M.run, entry.definition, input, profiles) + if not ok then + return "Error: " .. tostring(results) + end + 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. +function M.handle(input, profiles) + 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 + + 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) + 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) +end + +-- Discover the workflows and register a `/workflow:<name>` command for each +-- valid one. Invalid files register nothing; their errors stay in the +-- discovery warnings and surface through `subagents.workflow`. +function M.discover_and_register(profiles) + registry = M.discover() + local ext = host() + for _, entry in ipairs(registry.list) do + if entry.definition then + local def = entry.definition + ext.register_command({ + name = "workflow:" .. def.name, + description = def.description or ("Run the " .. def.name .. " workflow."), + handler = function(args) + local ok, results = pcall(M.run, def, args or "", profiles) + if not ok then + return "[workflow error: " .. tostring(results) .. "]" + end + return M.format_results(results) + end, + }) + end + end + return registry +end + +return M diff --git a/subagents/workflow.lua b/subagents/workflow.lua new file mode 100644 index 0000000..b1d3e46 --- /dev/null +++ b/subagents/workflow.lua @@ -0,0 +1,614 @@ +-- 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. +-- +-- ctx surface: +-- ctx:agent{ agent=, 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 +-- +-- Profile resolution and spawn-spec construction are NOT duplicated here: both +-- come from subagents/spawn.lua (`build_spec(input, profiles)` / `spawn(spec)`) +-- so the tool > profile > primary precedence lives in exactly one place. The +-- only thing this file adds to the spec is the synthetic structured-output +-- tool, which it normalizes to { name, description, schema } (defaulting the +-- name to "emit_result") so the host seam always sees the same shape. +-- +-- Edge cases and deliberate policies: +-- +-- * Child failures are values, never errors. A rejected spawn produces a +-- 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. +-- * "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 +-- settled result is ever lost between awaits. +-- * Handles the callback never awaited are awaited ("all") after it returns, +-- purely so no child is orphaned; those results are discarded. +-- * Structured output is decoded from result.structured_json and validated +-- against output.schema. Validation prefers the `jsonschema` rock and falls +-- back to the small built-in subset validator below when it is absent -- +-- that rock pulls in lrexlib-pcre, which needs a system PCRE and fails to +-- build on stock macOS, and a failed rock install would otherwise take the +-- whole extension down silently. A validation failure turns the result into +-- status "failed"; it is never reported as a successful structured result. +-- * The host seam is reached through `require("panto").ext` at call time, not +-- aliased at load time, matching subagents/spawn.lua so a test can install a +-- fake `panto` module before the first call rather than before the require. +-- subagents.jobs is required the same way, so load order between the two +-- never matters. + +local spawn = require("subagents.spawn") + +local M = {} + +local workflow_mt = { __name = "subagents.workflow" } + +M.workflow_mt = workflow_mt + +-- --------------------------------------------------------------------------- +-- Host seam access +-- --------------------------------------------------------------------------- + +local function host() + return require("panto").ext +end + +-- The job machinery, resolved at call time for the same reason as the host seam. +local function jobs() + return require("subagents.jobs") +end + +local function host_json() + local ok, ext = pcall(host) + if ok and type(ext) == "table" and type(ext.json) == "table" then + return ext.json + end + return nil +end + +-- Decode a JSON document. Panto installs its own codec as `panto.ext.json`; +-- the dkjson fallback only matters for a bare `lua` process running the specs. +local function json_decode(text) + local json = host_json() + if json and json.decode then + return json.decode(text) + end + local ok, dkjson = pcall(require, "dkjson") + if ok and type(dkjson) == "table" and dkjson.decode then + local value, _, err = dkjson.decode(text) + if err then + error(err, 0) + end + return value + end + error("no JSON decoder available (panto.ext.json missing, dkjson not installed)", 0) +end + +local function json_encode(value) + local json = host_json() + if json and json.encode then + local ok, encoded = pcall(json.encode, value) + if ok then + return encoded + end + end + local ok, dkjson = pcall(require, "dkjson") + if ok and type(dkjson) == "table" and dkjson.encode then + local encoded_ok, encoded = pcall(dkjson.encode, value) + if encoded_ok then + return encoded + end + end + return tostring(value) +end + +M.json_decode = json_decode +M.json_encode = json_encode + +-- --------------------------------------------------------------------------- +-- Schema validation +-- --------------------------------------------------------------------------- + +-- Built-in fallback validator: the JSON Schema subset that structured child +-- output actually uses. Anything it does not understand is ignored rather than +-- rejected, so an unrecognized keyword never fails a legitimate result. +local function is_array_like(value) + local count = 0 + for key in pairs(value) do + if type(key) ~= "number" then + return false + end + count = count + 1 + end + return count == #value +end + +local function type_matches(value, expected) + if expected == "object" then + return type(value) == "table" + elseif expected == "array" then + return type(value) == "table" and is_array_like(value) + elseif expected == "string" then + return type(value) == "string" + elseif expected == "number" then + return type(value) == "number" + elseif expected == "integer" then + return type(value) == "number" and value == math.floor(value) + elseif expected == "boolean" then + return type(value) == "boolean" + elseif expected == "null" then + return value == nil or type(value) == "userdata" + end + return true +end + +local function check_schema(value, schema, path) + if type(schema) ~= "table" then + return true + end + + local expected = schema.type + if type(expected) == "string" then + if not type_matches(value, expected) then + return false, string.format("%s: expected %s, got %s", path, expected, type(value)) + end + elseif type(expected) == "table" then + local any = false + for _, candidate in ipairs(expected) do + if type_matches(value, candidate) then + any = true + break + end + end + if not any then + return false, string.format("%s: no listed type matched %s", path, type(value)) + end + end + + if type(schema.enum) == "table" then + local found = false + for _, allowed in ipairs(schema.enum) do + if allowed == value then + found = true + break + end + end + if not found then + return false, string.format("%s: value is not one of the enumerated options", path) + end + end + + if type(value) == "string" then + if type(schema.minLength) == "number" and #value < schema.minLength then + return false, string.format("%s: shorter than minLength %d", path, schema.minLength) + end + if type(schema.maxLength) == "number" and #value > schema.maxLength then + return false, string.format("%s: longer than maxLength %d", path, schema.maxLength) + end + end + + if type(value) == "number" then + if type(schema.minimum) == "number" and value < schema.minimum then + return false, string.format("%s: below minimum %s", path, tostring(schema.minimum)) + end + if type(schema.maximum) == "number" and value > schema.maximum then + return false, string.format("%s: above maximum %s", path, tostring(schema.maximum)) + end + end + + if type(value) ~= "table" then + return true + end + + if type(schema.required) == "table" then + for _, key in ipairs(schema.required) do + if value[key] == nil then + return false, string.format("%s: missing required property '%s'", path, tostring(key)) + end + end + end + + if type(schema.properties) == "table" then + for key, sub in pairs(schema.properties) do + if value[key] ~= nil then + local ok, err = check_schema(value[key], sub, path .. "." .. tostring(key)) + if not ok then + return false, err + end + end + end + if schema.additionalProperties == false then + for key in pairs(value) do + if schema.properties[key] == nil then + return false, string.format("%s: unexpected property '%s'", path, tostring(key)) + end + end + end + end + + if type(schema.items) == "table" then + if type(schema.minItems) == "number" and #value < schema.minItems then + return false, string.format("%s: fewer than minItems %d", path, schema.minItems) + end + if type(schema.maxItems) == "number" and #value > schema.maxItems then + return false, string.format("%s: more than maxItems %d", path, schema.maxItems) + end + for index, item in ipairs(value) do + local ok, err = check_schema(item, schema.items, string.format("%s[%d]", path, index)) + if not ok then + return false, err + end + end + end + + return true +end + +local function validator_for(schema) + local ok, jsonschema = pcall(require, "jsonschema") + if ok and type(jsonschema) == "table" and jsonschema.generate_validator then + local generated_ok, generated = pcall(jsonschema.generate_validator, schema) + if generated_ok and type(generated) == "function" then + return generated + end + end + return function(value) + return check_schema(value, schema, "output") + end +end + +M.validate = function(value, schema) + return validator_for(schema)(value) +end + +-- --------------------------------------------------------------------------- +-- Result shaping +-- --------------------------------------------------------------------------- + +local function copy_result(result) + local shaped = {} + if type(result) == "table" then + for key, value in pairs(result) do + shaped[key] = value + end + end + if shaped.status == nil then + shaped.status = "failed" + shaped.error = shaped.error or "the host returned no result for this job" + shaped.resumable = false + end + return shaped +end + +local function fail(shaped, message) + shaped.status = "failed" + shaped.error = message + shaped.output = nil + return shaped +end + +-- Turn a host result into the value a workflow callback sees. Only handles +-- carrying an output schema decode structured JSON; everything else passes +-- through untouched. +local function shape_result(result, handle) + local shaped = copy_result(result) + local schema = handle and handle.output_schema + if schema == nil or shaped.status ~= "completed" then + return shaped + end + + local raw = shaped.structured_json + if type(raw) ~= "string" or raw == "" then + return fail(shaped, "structured output missing: the child produced no structured result") + end + + local decoded_ok, decoded = pcall(json_decode, raw) + if not decoded_ok then + return fail(shaped, "structured output failed validation: " .. tostring(decoded)) + end + + local valid, message = validator_for(schema)(decoded) + if not valid then + return fail(shaped, "structured output failed validation: " .. tostring(message or "schema mismatch")) + end + + shaped.output = decoded + return shaped +end + +-- The host may hand back a single result table or an array of them; both are +-- normalized to an array here. A result always carries `status`, which is what +-- distinguishes the two shapes. +local function as_result_array(value) + if type(value) ~= "table" then + return {} + end + if value.status ~= nil then + return { value } + end + return value +end + +-- --------------------------------------------------------------------------- +-- Handles and context +-- --------------------------------------------------------------------------- + +local handle_mt = {} +handle_mt.__index = handle_mt +handle_mt.__name = "subagents.handle" + +function handle_mt:await() + return self.ctx:await({ self }, "all")[1] +end + +local ctx_mt = {} +ctx_mt.__index = ctx_mt +ctx_mt.__name = "subagents.ctx" + +-- Per-run state lives here, not on ctx: the sandboxed guest holds the ctx table +-- and would otherwise be able to raise its own job cap (`ctx.max_jobs = nil`), +-- reset the counter, or read the profile set. ctx itself is an empty table +-- exposing only `agent` and `await`. Weak keys so a finished run is collectable. +local state = setmetatable({}, { __mode = "k" }) + +-- 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 +-- outside the job cap. The guest sees only `result` and `await`. +local job_of = setmetatable({}, { __mode = "k" }) + +function ctx_mt:agent(input) + if type(input) ~= "table" then + error("ctx:agent expects a table of { agent =, prompt =, ... }", 2) + end + local s = state[self] + if not s then + error("ctx:agent must be called on a workflow context (use ctx:agent{...})", 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 + + -- 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 + error(tostring(spec_err), 2) + end + + local output_schema = nil + if type(spec.output) == "table" then + output_schema = spec.output.schema + end + + local handle = setmetatable({ + ctx = self, + spec = spec, + output_schema = output_schema, + result = nil, + }, handle_mt) + + local job, job_err = spawn.spawn(spec) + if not job then + -- A rejected spawn is a child failure, not a workflow error. + handle.result = { + id = nil, + status = "failed", + error = tostring(job_err or "the host refused to start the child"), + resumable = false, + } + else + job_of[handle] = job + end + + s.job_count = s.job_count + 1 + s.outstanding[#s.outstanding + 1] = handle + return handle +end + +-- Pick the first handle (in input order) that already holds a settled result, +-- returning it with the remaining handles. +local function take_settled(handles) + for index, handle in ipairs(handles) do + if handle.result ~= nil then + local remaining = {} + for other_index, other in ipairs(handles) do + if other_index ~= index then + remaining[#remaining + 1] = other + end + end + return handle.result, remaining + end + end + return nil, nil +end + +local function pending_handles(handles) + local pending, started = {}, {} + for _, handle in ipairs(handles) do + if handle.result == nil and job_of[handle] ~= nil then + pending[#pending + 1] = handle + started[#started + 1] = job_of[handle] + end + end + return pending, started +end + +function ctx_mt:await(handles, mode) + mode = mode or "all" + if type(handles) ~= "table" then + error("ctx:await expects an array of handles", 2) + end + if getmetatable(handles) == handle_mt then + handles = { handles } + end + if mode ~= "all" and mode ~= "first" then + error("ctx:await mode must be \"all\" or \"first\"", 2) + end + + if mode == "all" then + local pending, started = pending_handles(handles) + 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) + end + end + local out = {} + for index, handle in ipairs(handles) do + out[index] = handle.result or copy_result(nil) + end + return out + end + + local ready, remaining = take_settled(handles) + if ready ~= nil then + return ready, remaining + end + + local pending, started = pending_handles(handles) + if #pending == 0 then + return nil, {} + end + + local results, still_pending = jobs().await(started, "first") + results = as_result_array(results) + + -- Everything not listed as still-running has settled; pair those handles + -- with the returned results in order. The listed jobs are the very handles + -- that went in, so they match by identity. + local settled = pending + if type(still_pending) == "table" and #still_pending > 0 then + local still_running = {} + for _, job in ipairs(still_pending) do + still_running[job] = true + end + settled = {} + for _, handle in ipairs(pending) do + if not still_running[job_of[handle]] then + settled[#settled + 1] = handle + end + end + end + for index, result in ipairs(results) do + local handle = settled[index] + if handle then + handle.result = shape_result(result, handle) + end + end + + ready, remaining = take_settled(handles) + if ready == nil then + -- 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) + return take_settled(handles) + end + return ready, remaining +end + +-- --------------------------------------------------------------------------- +-- Workflow objects and execution +-- --------------------------------------------------------------------------- + +function M.workflow(fn) + if type(fn) ~= "function" then + error("subagents.workflow expects a function(ctx, input)", 2) + end + return setmetatable({ run = fn }, workflow_mt) +end + +function M.is_workflow(value) + return type(value) == "table" and getmetatable(value) == workflow_mt +end + +-- Settle any handle the callback left running so a returning workflow never +-- orphans a child. Results are intentionally discarded. +local function drain(ctx) + local pending = {} + for _, handle in ipairs(state[ctx].outstanding) do + if handle.result == nil then + pending[#pending + 1] = handle + end + end + if #pending == 0 then + return + end + pcall(function() + ctx:await(pending, "all") + end) +end + +-- Run `wf` against `input`. opts: +-- max_jobs -- cap on ctx:agent calls (nil = unbounded; the sandbox passes 32) +-- profiles -- discovered profile set for spawn.build_spec (nil = discover) +-- 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. +function M.execute(wf, input, opts) + if not M.is_workflow(wf) then + error("subagents.workflow.execute expects a workflow object", 2) + end + opts = opts or {} + + local ctx = setmetatable({}, ctx_mt) + state[ctx] = { + profiles = opts.profiles, + max_jobs = opts.max_jobs, + job_count = 0, + outstanding = {}, + } + + local co = coroutine.running() + if opts.on_resume then + opts.on_resume(co) + end + -- pcall is yieldable in 5.4, so the callback may still await across it. + local packed = table.pack(pcall(wf.run, ctx, input)) + if opts.on_yield then + opts.on_yield(co) + end + + drain(ctx) + if not packed[1] then + error(packed[2], 0) + end + return table.unpack(packed, 2, packed.n) +end + +-- A child's output text: a structured result decodes to a table, which is +-- re-encoded compactly so anything model-visible is still a string. +function M.output_text(result) + local output = result.output + if type(output) == "table" then + return json_encode(output) + end + if output == nil or output == "" then + return tostring(result.error or "") + end + return tostring(output) +end + +return M |