diff --git a/src/exports.ts b/src/exports.ts index 514bc3a..8ef9bb9 100644 --- a/src/exports.ts +++ b/src/exports.ts @@ -86,10 +86,19 @@ export { getVoiceContext, getAgentContext } from "./context.js"; export { discoverSkills } from "./skills.js"; export { discoverWorkflows, + executeFlow, loadFlowDefinition, saveFlowDefinition, deleteFlowDefinition, } from "./workflows.js"; +export type { + FlowExecutionContext, + FlowExecutionResult, + FlowProgressEvent, + FlowStepResult, + SkillFlowDefinition, + SkillFlowStep, +} from "./workflows.js"; export { discoverSchedules, saveSchedule, diff --git a/src/workflows.ts b/src/workflows.ts index 03fa300..4327ebe 100644 --- a/src/workflows.ts +++ b/src/workflows.ts @@ -15,6 +15,69 @@ export interface SkillFlowDefinition { steps: SkillFlowStep[]; } +export type FlowProgressEvent = + | { type: "start"; flow: SkillFlowDefinition; input: string } + | { type: "step_start"; flow: SkillFlowDefinition; step: SkillFlowStep; index: number; input: string } + | { type: "step_complete"; flow: SkillFlowDefinition; step: SkillFlowStep; index: number; input: string; output: string } + | { type: "complete"; flow: SkillFlowDefinition; input: string; output: string; status: "completed" | "rejected" }; + +export interface FlowExecutionContext { + /** Execute one non-approval step. The input is the previous step's output. */ + runStep: (step: SkillFlowStep, input: string, index: number) => Promise; + /** Resolve an approval gate. Return false to stop the flow without running later steps. */ + requestApproval?: (step: SkillFlowStep, index: number) => Promise; + onProgress?: (event: FlowProgressEvent) => void; +} + +export interface FlowStepResult { + step: SkillFlowStep; + index: number; + input: string; + output: string; +} + +export interface FlowExecutionResult { + flow: SkillFlowDefinition; + input: string; + output: string; + status: "completed" | "rejected"; + steps: FlowStepResult[]; +} + +/** Execute a SkillFlow without assuming a transport or agent implementation. */ +export async function executeFlow( + flow: SkillFlowDefinition, + input: string, + context: FlowExecutionContext, +): Promise { + if (!flow.steps || flow.steps.length === 0) throw new Error("Flow must have at least one step"); + let currentInput = input; + const steps: FlowStepResult[] = []; + context.onProgress?.({ type: "start", flow, input }); + + for (const [index, step] of flow.steps.entries()) { + context.onProgress?.({ type: "step_start", flow, step, index, input: currentInput }); + if (step.skill === "__approval_gate__") { + if (!context.requestApproval) throw new Error(`Flow step ${index + 1} requires an approval callback`); + if (!(await context.requestApproval(step, index))) { + const result = { flow, input, output: currentInput, status: "rejected" as const, steps }; + context.onProgress?.({ type: "complete", ...result }); + return result; + } + continue; + } + const stepInput = currentInput; + const output = await context.runStep(step, stepInput, index); + steps.push({ step, index, input: stepInput, output }); + currentInput = output; + context.onProgress?.({ type: "step_complete", flow, step, index, input: stepInput, output }); + } + + const result = { flow, input, output: currentInput, status: "completed" as const, steps }; + context.onProgress?.({ type: "complete", ...result }); + return result; +} + export interface WorkflowMetadata { name: string; description: string; diff --git a/test/workflows.test.ts b/test/workflows.test.ts new file mode 100644 index 0000000..195f728 --- /dev/null +++ b/test/workflows.test.ts @@ -0,0 +1,57 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { executeFlow } from "../dist/workflows.js"; + +const flow = { + name: "daily-report", + description: "Build a report", + steps: [ + { skill: "collect", prompt: "Collect data" }, + { skill: "__approval_gate__", prompt: "Approve publication", channel: "telegram" }, + { skill: "publish", prompt: "Publish report" }, + ], +}; + +test("executeFlow chains outputs and emits progress", async () => { + const calls: string[] = []; + const events: string[] = []; + const result = await executeFlow(flow, "Monday", { + runStep: async (step, input, index) => { + calls.push(`${index}:${step.skill}:${input}`); + return `${input}->${step.skill}`; + }, + requestApproval: async (step, index) => { + calls.push(`${index}:${step.skill}`); + return true; + }, + onProgress: (event) => events.push(event.type), + }); + + assert.equal(result.status, "completed"); + assert.equal(result.output, "Monday->collect->publish"); + assert.deepEqual(calls, ["0:collect:Monday", "1:__approval_gate__", "2:publish:Monday->collect"]); + assert.deepEqual(events, ["start", "step_start", "step_complete", "step_start", "step_start", "step_complete", "complete"]); +}); + +test("executeFlow stops safely when an approval gate is rejected", async () => { + let runCount = 0; + const result = await executeFlow(flow, "input", { + runStep: async (_step, input) => { + runCount++; + return `${input}:collected`; + }, + requestApproval: async () => false, + }); + + assert.equal(result.status, "rejected"); + assert.equal(result.output, "input:collected"); + assert.equal(result.steps.length, 1); + assert.equal(runCount, 1); +}); + +test("executeFlow requires an approval callback for approval gates", async () => { + await assert.rejects( + executeFlow(flow, "input", { runStep: async (_step, input) => input }), + /requires an approval callback/, + ); +});