feat(workflows): expose transport-agnostic flow executor

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,

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<string>;
+	/** Resolve an approval gate. Return false to stop the flow without running later steps. */
+	requestApproval?: (step: SkillFlowStep, index: number) => Promise<boolean>;
+	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<FlowExecutionResult> {
+	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;

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/,
+	);
+});