diff options
Diffstat (limited to 'spec/test_workflow.lua')
| -rw-r--r-- | spec/test_workflow.lua | 299 |
1 files changed, 299 insertions, 0 deletions
diff --git a/spec/test_workflow.lua b/spec/test_workflow.lua new file mode 100644 index 0000000..c5ba06c --- /dev/null +++ b/spec/test_workflow.lua @@ -0,0 +1,299 @@ +-- subagents/workflow.lua: the callback API's awaiting, failure-as-value rule, +-- structured workers, the concurrency gate, and the job budget. +-- +-- The fake host settles jobs in the order each outcome's `settle` key asks for, +-- which is what makes the ordering cases meaningful: input order and settle +-- order are deliberately different everywhere below. + +local fake = require("spec.fake_ext") +local workflow = require("subagents.workflow") + +local function has(text, needle) + assert(type(text) == "string", "expected a string, got " .. type(text)) + assert(text:find(needle, 1, true), "expected to find " .. needle .. " in:\n" .. tostring(text)) +end + +local function profile_set() + local set = { list = {}, by_name = {}, warnings = {} } + for _, name in ipairs({ "alpha", "beta", "gamma" }) do + local profile = { name = name, description = name, body = "You are " .. name .. ".\n" } + set.list[#set.list + 1] = profile + set.by_name[name] = profile + end + return set +end + +local function with_host(fn, opts) + local handle = fake.install(opts) + local ok, err = pcall(fn, handle, profile_set()) + handle.restore() + if not ok then + error(err, 0) + end +end + +local function three_children(ctx) + local handles = {} + for index, name in ipairs({ "alpha", "beta", "gamma" }) do + handles[index] = ctx:agent({ agent = name, prompt = "work on " .. name }) + end + return handles +end + +local function json_available() + return pcall(require, "dkjson") +end + +local ITEMS_SCHEMA = { + type = "object", + required = { "items" }, + properties = { + items = { type = "array", items = { type = "string" } }, + }, +} + +local GHOST = { unknown_models = { ["openai:ghost"] = true } } + +return { + { "await all returns results in input order, not settle order", function() + with_host(function(handle, profiles) + handle.queue_for("alpha", { output = "A", settle = 3 }) + handle.queue_for("beta", { output = "B", settle = 1 }) + handle.queue_for("gamma", { output = "C", settle = 2 }) + + local results = workflow.execute(workflow.workflow(function(ctx) + return ctx:await(three_children(ctx), "all") + end), "input", { profiles = profiles }) + + assert(#results == 3, "expected three results") + assert(results[1].output == "A", tostring(results[1].output)) + assert(results[2].output == "B", tostring(results[2].output)) + assert(results[3].output == "C", tostring(results[3].output)) + assert(#handle.spawns == 3, "one child per ctx:agent") + assert(handle.spawns[1].prompt == "work on alpha") + end) + end }, + + { "await first returns the earliest settler and the remaining handles", function() + with_host(function(handle, profiles) + handle.queue_for("alpha", { output = "A", settle = 3 }) + handle.queue_for("beta", { output = "B", settle = 1 }) + handle.queue_for("gamma", { output = "C", settle = 2 }) + + local order = workflow.execute(workflow.workflow(function(ctx) + local handles = three_children(ctx) + local seen = {} + while #handles > 0 do + local result, remaining = ctx:await(handles, "first") + seen[#seen + 1] = result.output + assert(#remaining == #handles - 1, "one handle settles per await") + handles = remaining + end + return seen + end), "input", { profiles = profiles }) + + assert(table.concat(order, ",") == "B,C,A", table.concat(order, ",")) + end) + end }, + + { "handle:await settles one child", function() + with_host(function(handle, profiles) + handle.queue_for("alpha", { output = "just me" }) + local result = workflow.execute(workflow.workflow(function(ctx) + return ctx:agent({ agent = "alpha", prompt = "go" }):await() + end), "input", { profiles = profiles }) + assert(result.status == "completed", tostring(result.status)) + assert(result.output == "just me", tostring(result.output)) + end) + end }, + + { "the gate keeps four children in flight and queues the rest", function() + with_host(function(handle, profiles) + local results = workflow.execute(workflow.workflow(function(ctx) + local handles = {} + for index = 1, 5 do + handles[index] = ctx:agent({ agent = "alpha", prompt = "task " .. index }) + end + assert(#handle.runs == 4, "four turns run at once, saw " .. #handle.runs) + return ctx:await(handles, "all") + end), "input", { profiles = profiles }) + + assert(#results == 5, "every child reports") + assert(#handle.runs == 5, "the queued child starts once a slot frees up") + assert(handle.max_live == 4, "never more than four at once, peaked at " .. handle.max_live) + end) + end }, + + { "a rejected child is a failed result, not an error", function() + with_host(function(handle, profiles) + handle.queue_for("beta", { output = "B" }) + + local results = workflow.execute(workflow.workflow(function(ctx) + local first = ctx:agent({ agent = "alpha", model = "openai:ghost", prompt = "a" }) + local second = ctx:agent({ agent = "beta", prompt = "b" }) + return ctx:await({ first, second }, "all") + end), "input", { profiles = profiles }) + + assert(results[1].status == "failed", tostring(results[1].status)) + has(results[1].error, "unknown model 'openai:ghost'") + assert(results[1].resumable == false, "a child that never allocated is not resumable") + assert(results[2].status == "completed", "a sibling failure must not disturb this one") + assert(#handle.runs == 1, "only the sibling was ever started") + end, GHOST) + end }, + + { "a failed child is reported as a value", function() + with_host(function(handle, profiles) + handle.queue_for("alpha", { status = "failed", error = "provider refused", resumable = false }) + local result = workflow.execute(workflow.workflow(function(ctx) + return ctx:agent({ agent = "alpha", prompt = "a" }):await() + end), "input", { profiles = profiles }) + assert(result.status == "failed") + assert(result.error == "provider refused", tostring(result.error)) + end) + end }, + + { "an id with a turn already in flight is refused", function() + with_host(function(handle, profiles) + handle.add_session("0198-child", { + { role = "system", text = "You are alpha.\n", + metadata = { subagents = { owner = "0198-primary", agent = "alpha" } } }, + }) + handle.queue({ output = "the first turn wins" }) + + local results = workflow.execute(workflow.workflow(function(ctx) + local first = ctx:agent({ id = "0198-child", prompt = "a" }) + local second = ctx:agent({ id = "0198-child", prompt = "b" }) + return ctx:await({ first, second }, "all") + end), "input", { profiles = profiles }) + + assert(results[1].status == "completed", tostring(results[1].error)) + assert(results[2].status == "failed", "one turn per child at a time") + has(results[2].error, "already has a turn in flight") + assert(#handle.runs == 1, "the second call never starts a turn") + end) + end }, + + { "a structured worker decodes and validates its output", function() + if not json_available() then + return "skip", "dkjson is not installed" + end + with_host(function(handle, profiles) + handle.queue_for("alpha", { structured_json = '{"items":["x","y"]}' }) + local result = workflow.execute(workflow.workflow(function(ctx) + return ctx:agent({ + agent = "alpha", + prompt = "split it", + output = { description = "Return the work items.", schema = ITEMS_SCHEMA }, + }):await() + end), "input", { profiles = profiles }) + + assert(result.status == "completed", tostring(result.error)) + assert(type(result.output) == "table", "structured output decodes to a table") + assert(result.output.items[1] == "x" and result.output.items[2] == "y") + assert(result.resumable == false, "a one-shot child is not resumable") + assert(result.id == nil, "and has no durable session to name") + assert(#handle.stores == 0, "a structured worker never touches the child catalog") + + local child = handle.spawns[1] + assert(type(child.tool_choice) == "table" and child.tool_choice.name == "emit_result", + "the output tool is the only choice the child has") + assert(child.run.dispatch_tools == false, "a one-shot child never dispatches a tool call") + assert(#child.tool_decls == 1, "the output tool replaces the inherited set, saw " .. + table.concat(child.tools, ",")) + + local decl = child.tool_decls[1] + assert(decl.name == "emit_result", "the synthetic tool is named for the host") + assert(decl.description == "Return the work items.") + assert(decl.schema == ITEMS_SCHEMA, "the schema is passed through untouched") + assert(decl._source == nil and decl._ctx == nil and decl._vt == nil, + "the output tool is declaration-only: nothing may dispatch it") + end) + end }, + + { "structured output that violates the schema fails the result", function() + if not json_available() then + return "skip", "dkjson is not installed" + end + with_host(function(handle, profiles) + handle.queue_for("alpha", { structured_json = '{"nope":1}' }) + local result = workflow.execute(workflow.workflow(function(ctx) + return ctx:agent({ + agent = "alpha", + prompt = "split it", + output = { schema = ITEMS_SCHEMA }, + }):await() + end), "input", { profiles = profiles }) + + assert(result.status == "failed", "a schema violation is never a success") + has(result.error, "structured output failed validation") + assert(result.output == nil, "invalid output is not handed to the caller") + end) + end }, + + { "a structured worker that answers in prose fails", function() + with_host(function(handle, profiles) + handle.queue_for("alpha", { output = "prose, not a tool call" }) + local result = workflow.execute(workflow.workflow(function(ctx) + return ctx:agent({ + agent = "alpha", + prompt = "split it", + output = { schema = ITEMS_SCHEMA }, + }):await() + end), "input", { profiles = profiles }) + assert(result.status == "failed", tostring(result.status)) + has(result.error, "did not call the required 'emit_result' output tool") + end) + end }, + + { "an already settled handle is served without polling again", function() + with_host(function(handle, profiles) + handle.queue_for("beta", { output = "B" }) + handle.queue_for("gamma", { output = "C" }) + + local first_status = workflow.execute(workflow.workflow(function(ctx) + local rejected = ctx:agent({ agent = "alpha", model = "openai:ghost", prompt = "a" }) + local live = { rejected, ctx:agent({ agent = "beta", prompt = "b" }) } + local result, remaining = ctx:await(live, "first") + assert(#remaining == 1, "the pending sibling stays outstanding") + assert(handle.polls == 0, "a cached result must not reach the job machinery") + return result.status + end), "input", { profiles = profiles }) + + assert(first_status == "failed", tostring(first_status)) + assert(#handle.runs == 1, "only the live sibling was ever started") + end, GHOST) + end }, + + { "max_jobs caps how many children one workflow starts", function() + with_host(function(handle, profiles) + local ok, err = pcall(workflow.execute, workflow.workflow(function(ctx) + three_children(ctx) + end), "input", { profiles = profiles, max_jobs = 2 }) + assert(not ok, "the third ctx:agent must be refused") + has(tostring(err), "job limit exceeded") + assert(#handle.runs == 2, "the capped call never reaches the host") + end) + end }, + + { "handles the callback never awaited are settled before returning", function() + with_host(function(handle, profiles) + workflow.execute(workflow.workflow(function(ctx) + ctx:agent({ agent = "alpha", prompt = "orphan" }) + return "done" + end), "input", { profiles = profiles }) + assert(#handle.jobs == 1, "one child ran") + assert(handle.jobs[1]._settled, "no child is left running") + end) + end }, + + { "an unknown agent inside a workflow is a workflow error", function() + with_host(function(handle, profiles) + local ok, err = pcall(workflow.execute, workflow.workflow(function(ctx) + ctx:agent({ agent = "ghost", prompt = "go" }) + end), "input", { profiles = profiles }) + assert(not ok, "an unknown profile is a programmer error, not a child failure") + has(tostring(err), "unknown agent 'ghost'") + end) + end }, +} |