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.

1 files changed, 614 insertions, 0 deletions

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