feat: run SkillFlows from the CLI, SDK and scheduler

src/exports.ts

@@ -90,6 +90,15 @@ export {
 	saveFlowDefinition,
 	deleteFlowDefinition,
 } from "./workflows.js";
+export type { SkillFlowStep, SkillFlowDefinition } from "./workflows.js";
+// Flow execution lives in core so the CLI, scheduler, SDK and voice all run
+// SkillFlows through the same loop; callers inject their own side effects.
+export { executeFlow, APPROVAL_GATE } from "./flow-runner.js";
+export type { FlowRunDeps, FlowEvent, FlowResult, FlowStepResult } from "./flow-runner.js";
+// The batteries-included form: runFlow supplies the default "run this step in
+// this agent" behaviour so callers don't hand-write it.
+export { runFlow } from "./run-flow.js";
+export type { RunFlowOptions, RunFlowResult, ApprovalPolicy, StepUsage } from "./run-flow.js";
 export {
 	discoverSchedules,
 	saveSchedule,

src/flow-runner.ts

@@ -0,0 +1,132 @@
+// Executes a SkillFlow: runs each step in order, threading every step's output
+// into the next one's prompt. All side effects are injected (RunFlowDeps) so the
+// same loop drives the CLI, the scheduler, the SDK, and the voice web UI —
+// the caller supplies how to run a step, how to report progress, and how to ask
+// a human for approval.
+
+import type { SkillFlowDefinition, SkillFlowStep } from "./workflows.js";
+
+/** Reserved pseudo-skill: pauses the flow until a human approves. */
+export const APPROVAL_GATE = "__approval_gate__";
+
+export type FlowEvent =
+	| { type: "flow_start"; flow: string; totalSteps: number }
+	| { type: "step_start"; index: number; total: number; skill: string }
+	| { type: "step_done"; index: number; total: number; skill: string; output: string }
+	| { type: "approval_requested"; index: number; total: number; channel?: string }
+	| { type: "approval_resolved"; index: number; approved: boolean }
+	| { type: "flow_done"; flow: string; steps: number }
+	| { type: "flow_aborted"; flow: string; index: number; reason: string };
+
+export interface FlowRunDeps {
+	/**
+	 * Run one step and return the assistant's text. Receives the fully-built
+	 * prompt plus the raw step and its index — so a caller can inspect the step
+	 * for per-step settings (e.g. a future `model` field) without the loop
+	 * needing to know about them.
+	 */
+	runStep: (prompt: string, step: SkillFlowStep, index: number) => Promise<string>;
+	/** Progress reporting — terminal output, WebSocket frames, log lines. */
+	onProgress?: (event: FlowEvent) => void;
+	/**
+	 * Ask a human to approve continuing. Omit it and approval gates DENY:
+	 * a gate exists to guard a risky step, so with nobody to ask, the safe
+	 * answer is to stop. Callers wanting unattended runs must opt in explicitly
+	 * by passing `async () => true`.
+	 */
+	requestApproval?: (message: string, channel?: string) => Promise<boolean>;
+}
+
+export interface FlowStepResult {
+	index: number;
+	skill: string;
+	output: string;
+}
+
+export interface FlowResult {
+	flow: string;
+	/** False when an approval gate denied, timed out, or had no handler. */
+	completed: boolean;
+	steps: FlowStepResult[];
+	/** The accumulated context after the last step that ran. */
+	context: string;
+	/** 0-based index of the step that stopped the flow, when not completed. */
+	abortedAt?: number;
+	abortReason?: string;
+}
+
+function buildStepPrompt(step: SkillFlowStep, userContext: string, runningContext: string): string {
+	return `Use the skill "${step.skill}" (load it with /skill:${step.skill}).
+${step.prompt.replace(/\{input\}/g, userContext)}
+
+Context from previous steps:
+${runningContext}`;
+}
+
+function buildApprovalMessage(
+	flow: SkillFlowDefinition,
+	step: SkillFlowStep,
+	index: number,
+	runningContext: string,
+): string {
+	if (step.prompt) {
+		return `⏸ Approval Required: ${step.prompt}\n\nReply YES to continue or NO to cancel.`;
+	}
+	return `⏸ Flow "${flow.name}" paused at step ${index + 1}/${flow.steps.length}.\n\nCompleted so far:\n${runningContext.slice(0, 500)}\n\nReply YES to continue or NO to cancel.`;
+}
+
+/**
+ * Run a SkillFlow to completion, or stop at the first denied approval gate.
+ * Never throws on a denied gate — that's a normal outcome, reported via
+ * `completed: false`. Errors from `runStep` propagate to the caller.
+ */
+export async function executeFlow(
+	flow: SkillFlowDefinition,
+	userContext: string,
+	deps: FlowRunDeps,
+): Promise<FlowResult> {
+	const total = flow.steps.length;
+	const results: FlowStepResult[] = [];
+	let runningContext = userContext;
+
+	deps.onProgress?.({ type: "flow_start", flow: flow.name, totalSteps: total });
+
+	for (let i = 0; i < total; i++) {
+		const step = flow.steps[i];
+
+		if (step.skill === APPROVAL_GATE) {
+			const channel = step.channel;
+			deps.onProgress?.({ type: "approval_requested", index: i, total, channel });
+
+			// No handler → deny. See RunFlowDeps.requestApproval.
+			const approved = deps.requestApproval
+				? await deps.requestApproval(buildApprovalMessage(flow, step, i, runningContext), channel)
+				: false;
+
+			deps.onProgress?.({ type: "approval_resolved", index: i, approved });
+
+			if (!approved) {
+				const reason = deps.requestApproval
+					? `Approval denied at step ${i + 1}/${total}`
+					: `Approval gate at step ${i + 1}/${total} could not be delivered — no approval handler available`;
+				deps.onProgress?.({ type: "flow_aborted", flow: flow.name, index: i, reason });
+				return { flow: flow.name, completed: false, steps: results, context: runningContext, abortedAt: i, abortReason: reason };
+			}
+
+			runningContext += `\n\n[Step ${i + 1}: approval gate]: Approved${channel ? ` via ${channel}` : ""}`;
+			continue;
+		}
+
+		deps.onProgress?.({ type: "step_start", index: i, total, skill: step.skill });
+
+		const output = await deps.runStep(buildStepPrompt(step, userContext, runningContext), step, i);
+
+		results.push({ index: i, skill: step.skill, output });
+		runningContext += `\n\n[Step ${i + 1} result (${step.skill})]: ${output}`;
+
+		deps.onProgress?.({ type: "step_done", index: i, total, skill: step.skill, output });
+	}
+
+	deps.onProgress?.({ type: "flow_done", flow: flow.name, steps: results.length });
+	return { flow: flow.name, completed: true, steps: results, context: runningContext };
+}

src/index.ts

@@ -8,6 +8,9 @@ import { createBuiltinTools } from "./tools/index.js";
 import { createSandboxContext } from "./sandbox.js";
 import type { SandboxContext, SandboxConfig } from "./sandbox.js";
 import { expandSkillCommand, refreshSkills } from "./skills.js";
+import { executeFlow } from "./flow-runner.js";
+import { loadFlowDefinition, discoverWorkflows } from "./workflows.js";
+import { query } from "./sdk.js";
 import { loadHooksConfig, runHooks, wrapToolWithHooks } from "./hooks.js";
 import type { HooksConfig } from "./hooks.js";
 import { loadDeclarativeTools } from "./tool-loader.js";
@@ -638,7 +641,7 @@ async function main(): Promise<void> {
 	if (loaded.plugins.length > 0) {
 		console.log(dim(`Plugins: ${loaded.plugins.map((p) => p.manifest.id).join(", ")}`));
 	}
-	console.log(dim('Type /skills to list skills, /plugins to list plugins, /memory to view memory, /quit to exit\n'));
+	console.log(dim('Type /skills to list skills, /flows to list workflows, /plugins to list plugins, /memory to view memory, /quit to exit\n'));
 
 	// Single-shot mode
 	if (prompt) {
@@ -680,6 +683,59 @@ async function main(): Promise<void> {
 		output: process.stdout,
 	});
 
+	// Approval gates prompt on the same readline instance. `ask()` is awaiting
+	// the flow at this point, so the prompt is free.
+	const askApproval = (message: string): Promise<boolean> =>
+		new Promise((res) => {
+			console.log(`\n${message}`);
+			rl.question(green("approve? [y/N] "), (answer) => {
+				res(/^(y|yes|approve|ok|continue|go|proceed)$/i.test(answer.trim()));
+			});
+		});
+
+	// Run a SkillFlow. Each step is an isolated query() — matching the web UI —
+	// so steps don't inherit the REPL's conversation history, and a step only
+	// sees what the flow explicitly threads into its prompt.
+	const runFlowInRepl = async (flowName: string, userContext: string): Promise<void> => {
+		// Re-discover rather than using the startup snapshot, so a flow created
+		// while the REPL is open (e.g. in voice's builder) is visible. Same
+		// reason /skills calls refreshSkills.
+		const workflows = await discoverWorkflows(dir);
+		const meta = workflows.find((w) => w.name === flowName && w.type === "flow");
+		if (!meta) {
+			console.error(red(`Unknown flow: ${flowName}`));
+			return;
+		}
+
+		const flowDef = await loadFlowDefinition(join(agentDir, meta.filePath));
+
+		await executeFlow(flowDef, userContext, {
+			runStep: async (stepPrompt) => {
+				let out = "";
+				const result = query({
+					prompt: stepPrompt,
+					dir: agentDir,
+					model,
+					env,
+				});
+				for await (const msg of result) {
+					if (msg.type === "delta" && msg.deltaType === "text") process.stdout.write(msg.content);
+					if (msg.type === "assistant" && msg.content) out += msg.content;
+					if (msg.type === "tool_use") console.log(dim(`\n  · ${msg.toolName}`));
+				}
+				process.stdout.write("\n");
+				return out;
+			},
+			requestApproval: askApproval,
+			onProgress: (e) => {
+				if (e.type === "flow_start") console.log(dim(`▶ flow: ${e.flow} (${e.totalSteps} steps)`));
+				if (e.type === "step_start") console.log(dim(`\n▶ step ${e.index + 1}/${e.total}: ${e.skill}`));
+				if (e.type === "flow_done") console.log(dim(`✓ flow complete (${e.steps} steps)`));
+				if (e.type === "flow_aborted") console.log(red(`✗ ${e.reason}`));
+			},
+		});
+	};
+
 	const ask = (): void => {
 		rl.question(green("→ "), async (input) => {
 			const trimmed = input.trim();
@@ -733,6 +789,23 @@ async function main(): Promise<void> {
 				return;
 			}
 
+			if (trimmed === "/flows") {
+				const currentWorkflows = await discoverWorkflows(dir);
+				if (currentWorkflows.length === 0) {
+					console.log(dim("No workflows defined."));
+				} else {
+					for (const w of currentWorkflows) {
+						const runnable = w.type === "flow";
+						const label = runnable ? bold(`@${w.name}`) : bold(w.name);
+						const note = runnable ? dim(` [${w.steps?.length ?? 0} steps]`) : dim(" [reference only]");
+						console.log(`  ${label} — ${dim(w.description)}${note}`);
+					}
+					console.log(dim("\nRun a flow by typing @flow-name, optionally followed by input."));
+				}
+				ask();
+				return;
+			}
+
 			if (trimmed === "/tasks") {
 				try {
 					const tasksRaw = await readFile(join(gitagentDir, "learning", "tasks.json"), "utf-8");
@@ -788,6 +861,21 @@ async function main(): Promise<void> {
 				return;
 			}
 
+			// SkillFlow trigger: @flow-name [input]. Anchored to the start of the
+			// line so an @mention or email address mid-sentence can't fire a flow,
+			// and the name must be followed by whitespace or end-of-line — a bare
+			// \b would let "@daily-report.com" fire the flow with input ".com".
+			const flowMatch = trimmed.match(/^@([a-z0-9]+(?:-[a-z0-9]+)*)(?=\s|$)\s*([\s\S]*)$/);
+			if (flowMatch) {
+				try {
+					await runFlowInRepl(flowMatch[1], flowMatch[2].trim());
+				} catch (err: any) {
+					console.error(red(`Flow error: ${err.message}`));
+				}
+				ask();
+				return;
+			}
+
 			// Skill expansion: /skill:name [args]
 			let promptText = trimmed;
 			if (trimmed.startsWith("/skill:")) {