From 4f0a91ef55fe96835172bdad34feec1e2a0a0977 Mon Sep 17 00:00:00 2001 From: t Date: Sun, 16 Aug 2026 20:42:43 -0600 Subject: subagents extension on the generic host surfaces The rock now owns all subagent policy on top of libpanto-lua's generic APIs: children are ordinary panto.agent instances over rock-constructed stores, started with agent:run_async and awaited by arming uv.new_poll on each job's wake_fd from the tool handler's coroutine. subagents/jobs.lua carries the session policy the host used to own: the concurrency gate (4 running, FIFO queue, cancel-while-queued never starts), the await contract (results in input order; "first" returns settled plus remaining by identity), and settle-time shaping. subagents/spawn.lua seeds new children (primary system context, child role, profile body with manifest metadata), resolves model/reasoning through panto.ext.resolve_model, filters subagents.* out of the inherited tool set via agent:set_tools, and reads resume defaults back from stored message metadata. One-shot structured workers are a null_store agent with a declaration-only output tool, tool_choice forced, dispatch_tools=false. subagents/progress.lua renders per-tool-entry cards through the component handle's invalidate seam; turn_interrupt cancels live children, turn_end closes them. Spec suite rewritten against fakes of the new surfaces (98 cases), including gate/queue/cancel bounds, resume-default extraction, one-shot capture via unresolved tool calls, tool filtering, and manifest seeding. --- subagents/workflow.lua | 614 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 614 insertions(+) create mode 100644 subagents/workflow.lua (limited to 'subagents/workflow.lua') 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 -- cgit v1.3