From 07cdcb3b47db9f7b4b52b2d834f70935e3261d25 Mon Sep 17 00:00:00 2001 From: Nivesh353 Date: Tue, 28 Jul 2026 12:00:56 +0530 Subject: [PATCH 1/2] feat: run SkillFlows from the CLI, SDK and scheduler MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SkillFlow definitions lived in core but the executor lived in @open-gitagent/voice, so the CLI, SDK and scheduler could create and list flows without ever being able to run one — while core's system prompt told every agent that "@flow_name" worked. Move the step loop into core as flow-runner.ts with its side effects injected, so each front-end supplies only what is genuinely specific to it: executeFlow(flow, input, { runStep, onProgress, requestApproval }) - CLI: "@flow-name [input]" trigger and a /flows command. The trigger is anchored to the start of the line so an address like bob@daily-report.com cannot fire a flow. Gates prompt on the existing readline instance. - Schedules: new optional `flow` field, so a schedule can run a SkillFlow instead of a single prompt. `prompt` becomes the flow's input. A flow schedule with no executor fails loudly. - SDK: runFlow() wraps executeFlow with the obvious defaults (run each step as an isolated query against the agent dir) and reports per-step token usage, so callers don't hand-write a runStep. Approval gates now deny when no approval handler is supplied. A gate guards a risky step; with nobody to ask, continuing silently removes the protection it exists to provide. Unattended callers opt in with `approve: "auto"`. Per-step usage counts cache reads and writes, which the session totals omit — a cache-heavy step otherwise reports near-zero input. Verified end to end: CLI approve/deny, SDK with a custom runStep, runFlow across all approval policies, and a scheduled flow. 21 new tests, none requiring an API key. --- src/exports.ts | 9 ++ src/flow-runner.ts | 132 ++++++++++++++++++++++++ src/index.ts | 88 +++++++++++++++- src/run-flow.ts | 157 ++++++++++++++++++++++++++++ src/schedule-runner.ts | 18 +++- src/schedules.ts | 23 +++-- test/flow-runner.test.ts | 216 +++++++++++++++++++++++++++++++++++++++ test/run-flow.test.ts | 144 ++++++++++++++++++++++++++ 8 files changed, 777 insertions(+), 10 deletions(-) create mode 100644 src/flow-runner.ts create mode 100644 src/run-flow.ts create mode 100644 test/flow-runner.test.ts create mode 100644 test/run-flow.test.ts diff --git a/src/exports.ts b/src/exports.ts index 514bc3a..51ba71a 100644 --- a/src/exports.ts +++ b/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, diff --git a/src/flow-runner.ts b/src/flow-runner.ts new file mode 100644 index 0000000..628a319 --- /dev/null +++ b/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; + /** 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; +} + +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 { + 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 }; +} diff --git a/src/index.ts b/src/index.ts index ba4eeb2..e01d40e 100644 --- a/src/index.ts +++ b/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 { 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 { 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 => + 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 => { + // 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 { 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,19 @@ async function main(): Promise { 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. + const flowMatch = trimmed.match(/^@([a-z0-9]+(?:-[a-z0-9]+)*)\b\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:")) { diff --git a/src/run-flow.ts b/src/run-flow.ts new file mode 100644 index 0000000..d3fba97 --- /dev/null +++ b/src/run-flow.ts @@ -0,0 +1,157 @@ +// Batteries-included wrapper over executeFlow. The core loop stays agnostic +// about how a step runs so the CLI and the web UI can plug in their own +// behaviour — but the common case ("just run this flow in this agent") has an +// obvious answer, and callers shouldn't have to hand-write it. This supplies +// that default: each step is an isolated query() against the agent directory. +// +// Anything unusual (streaming to a browser, a custom model per step, a human +// approver) drops down to executeFlow directly. + +import { join } from "path"; +import { executeFlow, type FlowEvent, type FlowResult } from "./flow-runner.js"; +import { discoverWorkflows, loadFlowDefinition, type SkillFlowDefinition } from "./workflows.js"; +import { query } from "./sdk.js"; + +/** + * What to do at an approval gate. + * - "deny" (default): gates stop the flow. Safe for unattended runs. + * - "auto": gates pass. Opt in deliberately — it disables the protection. + * - function: ask a human however you like; return true to continue. + */ +export type ApprovalPolicy = + | "deny" + | "auto" + | ((message: string, channel?: string) => Promise); + +export interface StepUsage { + index: number; + skill: string; + /** Fresh (uncached) input tokens. With prompt caching on, this is small. */ + inputTokens: number; + /** Input served from cache — often the bulk of a step's input. */ + cacheReadTokens: number; + /** Input written into the cache this step. */ + cacheWriteTokens: number; + outputTokens: number; + /** input + cacheRead + cacheWrite + output — the whole billable volume. */ + totalTokens: number; + requests: number; + costUsd: number; + /** Model ids that served this step, comma-joined. */ + models: string; +} + +export interface RunFlowOptions { + /** Agent directory — the one holding agent.yaml, skills/ and workflows/. */ + agentDir: string; + /** Flow name (e.g. "daily-report") or an already-loaded definition. */ + flow: string | SkillFlowDefinition; + /** Text substituted for {input} in step prompts. */ + input?: string; + /** Override the agent's model for every step. */ + model?: string; + /** Env profile passed through to the agent loader. */ + env?: string; + /** Approval gate policy. Defaults to "deny". */ + approve?: ApprovalPolicy; + /** Flow lifecycle events — same stream the CLI prints and the UI renders. */ + onProgress?: (event: FlowEvent) => void; + /** Token usage for each step, reported as soon as that step finishes. */ + onStepUsage?: (usage: StepUsage) => void; +} + +export interface RunFlowResult extends FlowResult { + /** Per-step token usage, in step order. */ + usage: StepUsage[]; + /** Sum of every step's cost. */ + totalCostUsd: number; + totalTokens: number; +} + +function resolveApproval(policy: ApprovalPolicy | undefined) { + if (typeof policy === "function") return policy; + if (policy === "auto") return async () => true; + // "deny" and undefined both mean: no handler, so executeFlow denies and + // says why. Returning undefined keeps that message accurate. + return undefined; +} + +async function resolveFlow(agentDir: string, flow: string | SkillFlowDefinition): Promise { + if (typeof flow !== "string") return flow; + + const workflows = await discoverWorkflows(agentDir); + const meta = workflows.find((w) => w.name === flow); + + if (!meta) { + const runnable = workflows.filter((w) => w.type === "flow").map((w) => w.name); + throw new Error( + `Flow "${flow}" not found in ${agentDir}/workflows` + + (runnable.length ? ` — available: ${runnable.join(", ")}` : " — no runnable flows defined"), + ); + } + if (meta.type !== "flow") { + throw new Error(`"${flow}" is a reference workflow (${meta.format}), not a runnable SkillFlow — it has no steps to execute`); + } + + return loadFlowDefinition(join(agentDir, meta.filePath)); +} + +/** + * Run a SkillFlow in an agent directory. Each step is an isolated agent call, + * with the previous steps' output threaded into its prompt. + * + * const r = await runFlow({ agentDir: "./my-agent", flow: "daily-report", input: "..." }); + * + * Approval gates deny by default — pass `approve: "auto"` or a function to + * change that. + */ +export async function runFlow(opts: RunFlowOptions): Promise { + const flow = await resolveFlow(opts.agentDir, opts.flow); + const usage: StepUsage[] = []; + + const result = await executeFlow(flow, opts.input ?? "", { + runStep: async (prompt, step, index) => { + const q = query({ prompt, dir: opts.agentDir, model: opts.model, env: opts.env }); + + let output = ""; + for await (const msg of q) { + if (msg.type === "assistant" && msg.content) output += msg.content; + } + + // One query per step, so this session's costs are this step's alone. + // Cache counters only exist per-model — the session totals cover + // fresh input and output only, so read them off modelUsage or a + // cache-heavy step looks nearly free. + const costs = q.costs(); + const models = Object.values(costs.modelUsage ?? {}); + const cacheReadTokens = models.reduce((n, m) => n + (m.cacheReadTokens ?? 0), 0); + const cacheWriteTokens = models.reduce((n, m) => n + (m.cacheWriteTokens ?? 0), 0); + + const entry: StepUsage = { + index, + skill: step.skill, + inputTokens: costs.totalInputTokens, + cacheReadTokens, + cacheWriteTokens, + outputTokens: costs.totalOutputTokens, + totalTokens: costs.totalInputTokens + cacheReadTokens + cacheWriteTokens + costs.totalOutputTokens, + requests: costs.totalRequests, + costUsd: costs.totalCostUsd, + models: Object.keys(costs.modelUsage ?? {}).join(",") || "-", + }; + usage.push(entry); + opts.onStepUsage?.(entry); + + return output; + }, + requestApproval: resolveApproval(opts.approve), + onProgress: opts.onProgress, + }); + + return { + ...result, + usage, + totalCostUsd: usage.reduce((n, u) => n + u.costUsd, 0), + totalTokens: usage.reduce((n, u) => n + u.totalTokens, 0), + }; +} diff --git a/src/schedule-runner.ts b/src/schedule-runner.ts index 9e677ce..e4862d2 100644 --- a/src/schedule-runner.ts +++ b/src/schedule-runner.ts @@ -11,6 +11,13 @@ export interface SchedulerOptions { model?: string; env?: string; runPrompt: (prompt: string) => Promise; + /** + * Run a SkillFlow by name, with the schedule's prompt as its input. Required + * for schedules that set `flow`. The caller owns the approval policy — pass + * a `requestApproval` through to `executeFlow` to allow unattended gates, + * otherwise gates deny and the job reports why. + */ + runFlow?: (flowName: string, input: string) => Promise; broadcastToBrowsers: (msg: ServerMessage) => void; appendToHistory: (msg: any) => void; } @@ -95,7 +102,7 @@ export async function executeScheduledJob(schedule: ScheduleDefinition, opts: Sc console.log(dim(`[scheduler] Running "${schedule.id}" at ${ts}`)); // Broadcast schedule start to chat - const startMsg = { type: "schedule_start", id: schedule.id, prompt: schedule.prompt, ts } as any; + const startMsg = { type: "schedule_start", id: schedule.id, prompt: schedule.prompt, ...(schedule.flow ? { flow: schedule.flow } : {}), ts } as any; opts.broadcastToBrowsers(startMsg as ServerMessage); opts.appendToHistory(startMsg); @@ -103,7 +110,14 @@ export async function executeScheduledJob(schedule: ScheduleDefinition, opts: Sc let success = true; try { - result = await opts.runPrompt(schedule.prompt); + if (schedule.flow) { + if (!opts.runFlow) { + throw new Error(`Schedule "${schedule.id}" runs flow "${schedule.flow}" but this runtime has no flow executor`); + } + result = await opts.runFlow(schedule.flow, schedule.prompt); + } else { + result = await opts.runPrompt(schedule.prompt); + } } catch (err: any) { result = err.message || "Unknown error"; success = false; diff --git a/src/schedules.ts b/src/schedules.ts index 44b2970..e8dbd5b 100644 --- a/src/schedules.ts +++ b/src/schedules.ts @@ -5,7 +5,10 @@ import yaml from "js-yaml"; export interface ScheduleDefinition { id: string; + /** Free-text prompt to run. When `flow` is set this is the flow's input instead. */ prompt: string; + /** Name of a SkillFlow to run instead of a plain prompt. */ + flow?: string; cron: string; mode: "repeat" | "once"; runAt?: string; // ISO datetime for "once" mode (alternative to cron) @@ -39,10 +42,11 @@ export async function discoverSchedules(agentDir: string): Promise; - if (data?.id && data?.prompt && (data?.cron || data?.runAt)) { + if (data?.id && (data?.prompt || data?.flow) && (data?.cron || data?.runAt)) { schedules.push({ id: String(data.id), - prompt: String(data.prompt), + prompt: String(data.prompt || ""), + ...(data.flow ? { flow: String(data.flow) } : {}), cron: String(data.cron || ""), mode: data.mode === "once" ? "once" : "repeat", ...(data.runAt ? { runAt: String(data.runAt) } : {}), @@ -63,12 +67,13 @@ export async function discoverSchedules(agentDir: string): Promise { const raw = await readFile(filePath, "utf-8"); const data = yaml.load(raw) as Record; - if (!data?.id || !data?.prompt || (!data?.cron && !data?.runAt)) { - throw new Error("Invalid schedule definition: missing id, prompt, or cron/runAt"); + if (!data?.id || (!data?.prompt && !data?.flow) || (!data?.cron && !data?.runAt)) { + throw new Error("Invalid schedule definition: missing id, prompt/flow, or cron/runAt"); } return { id: String(data.id), - prompt: String(data.prompt), + prompt: String(data.prompt || ""), + ...(data.flow ? { flow: String(data.flow) } : {}), cron: String(data.cron || ""), mode: data.mode === "once" ? "once" : "repeat", ...(data.runAt ? { runAt: String(data.runAt) } : {}), @@ -83,8 +88,11 @@ export async function saveSchedule(agentDir: string, schedule: ScheduleDefinitio if (!KEBAB_RE.test(schedule.id)) { throw new Error("Schedule id must be kebab-case (e.g. daily-standup)"); } - if (!schedule.prompt || (!schedule.cron && !schedule.runAt)) { - throw new Error("Schedule must have a prompt and cron expression or runAt time"); + if ((!schedule.prompt && !schedule.flow) || (!schedule.cron && !schedule.runAt)) { + throw new Error("Schedule must have a prompt or flow, and a cron expression or runAt time"); + } + if (schedule.flow && !KEBAB_RE.test(schedule.flow)) { + throw new Error("Schedule flow must be a kebab-case flow name (e.g. daily-report)"); } const schedulesDir = join(agentDir, "schedules"); mkdirSync(schedulesDir, { recursive: true }); @@ -92,6 +100,7 @@ export async function saveSchedule(agentDir: string, schedule: ScheduleDefinitio const content = yaml.dump({ id: schedule.id, prompt: schedule.prompt, + ...(schedule.flow ? { flow: schedule.flow } : {}), cron: schedule.cron || "", mode: schedule.mode || "repeat", ...(schedule.runAt ? { runAt: schedule.runAt } : {}), diff --git a/test/flow-runner.test.ts b/test/flow-runner.test.ts new file mode 100644 index 0000000..9ca6508 --- /dev/null +++ b/test/flow-runner.test.ts @@ -0,0 +1,216 @@ +// Tests for the core SkillFlow loop: step ordering, context threading, {input} +// substitution, approval-gate outcomes (approved / denied / no handler), and the +// progress event stream. `runStep` is injected, so the loop is exercised for real +// without spending an LLM call per assertion. + +import test from "node:test"; +import assert from "node:assert/strict"; + +import { executeFlow, APPROVAL_GATE, type FlowEvent, type FlowRunDeps } from "../src/flow-runner.ts"; +import type { SkillFlowDefinition } from "../src/workflows.ts"; + +function flow(...steps: SkillFlowDefinition["steps"]): SkillFlowDefinition { + return { name: "test-flow", description: "", steps }; +} + +/** Records every prompt runStep received and echoes a canned output per step. */ +function recorder(outputs: string[] = []) { + const prompts: string[] = []; + const seen: { skill: string; index: number }[] = []; + const deps: FlowRunDeps = { + runStep: async (prompt, step, index) => { + prompts.push(prompt); + seen.push({ skill: step.skill, index }); + return outputs[index] ?? `output-${index}`; + }, + }; + return { prompts, seen, deps }; +} + +// ── Ordering and context threading ───────────────────────────────────── + +test("runs every step in order and returns their outputs", async () => { + const { deps } = recorder(); + const r = await executeFlow( + flow( + { skill: "one", prompt: "do one" }, + { skill: "two", prompt: "do two" }, + { skill: "three", prompt: "do three" }, + ), + "user input", + deps, + ); + + assert.equal(r.completed, true); + assert.deepEqual(r.steps.map((s) => s.skill), ["one", "two", "three"]); + assert.deepEqual(r.steps.map((s) => s.output), ["output-0", "output-1", "output-2"]); +}); + +test("each step's prompt carries the previous steps' output forward", async () => { + const { prompts, deps } = recorder(["FIRST-RESULT", "SECOND-RESULT"]); + await executeFlow( + flow({ skill: "one", prompt: "do one" }, { skill: "two", prompt: "do two" }), + "seed", + deps, + ); + + // Step 1 only sees the user's input. + assert.match(prompts[0], /Context from previous steps:\nseed/); + assert.doesNotMatch(prompts[0], /FIRST-RESULT/); + + // Step 2 sees step 1's labelled result appended. + assert.match(prompts[1], /\[Step 1 result \(one\)\]: FIRST-RESULT/); + assert.match(prompts[1], /Use the skill "two"/); +}); + +test("{input} is replaced with the user's context", async () => { + const { prompts, deps } = recorder(); + await executeFlow(flow({ skill: "summarize", prompt: "Summarize {input} twice" }), "the Q3 report", deps); + assert.match(prompts[0], /Summarize the Q3 report twice/); +}); + +test("runStep receives the step itself and its index, not just the prompt", async () => { + // The loop stays agnostic about per-step settings — it hands the whole step + // to the caller, so fields added later (e.g. a per-step model) need no + // change here. Gates are not forwarded to runStep. + const { seen, deps } = recorder(); + await executeFlow( + flow( + { skill: "first", prompt: "x" }, + { skill: APPROVAL_GATE, prompt: "?" }, + { skill: "second", prompt: "y" }, + ), + "", + { ...deps, requestApproval: async () => true }, + ); + assert.deepEqual(seen, [ + { skill: "first", index: 0 }, + { skill: "second", index: 2 }, // index is the position in the flow, gates included + ]); +}); + +// ── Approval gates ───────────────────────────────────────────────────── + +test("approved gate lets the flow continue", async () => { + const { deps } = recorder(); + const r = await executeFlow( + flow( + { skill: "one", prompt: "do one" }, + { skill: APPROVAL_GATE, prompt: "ok to send?", channel: "telegram" }, + { skill: "two", prompt: "do two" }, + ), + "", + { ...deps, requestApproval: async () => true }, + ); + + assert.equal(r.completed, true); + assert.deepEqual(r.steps.map((s) => s.skill), ["one", "two"]); // gate is not a step result + assert.match(r.context, /approval gate\]: Approved via telegram/); +}); + +test("denied gate stops the flow and skips every later step", async () => { + const { prompts, deps } = recorder(); + const r = await executeFlow( + flow( + { skill: "one", prompt: "do one" }, + { skill: APPROVAL_GATE, prompt: "ok to delete?" }, + { skill: "destructive", prompt: "delete everything" }, + ), + "", + { ...deps, requestApproval: async () => false }, + ); + + assert.equal(r.completed, false); + assert.equal(r.abortedAt, 1); + assert.match(r.abortReason!, /denied at step 2\/3/); + assert.deepEqual(r.steps.map((s) => s.skill), ["one"]); + assert.equal(prompts.length, 1); // "destructive" never ran +}); + +test("gate with no approval handler DENIES rather than silently continuing", async () => { + const { prompts, deps } = recorder(); + const r = await executeFlow( + flow( + { skill: APPROVAL_GATE, prompt: "ok to send the email?" }, + { skill: "send-email", prompt: "send it" }, + ), + "", + deps, // no requestApproval + ); + + assert.equal(r.completed, false); + assert.equal(r.abortedAt, 0); + assert.match(r.abortReason!, /no approval handler available/); + assert.equal(prompts.length, 0); +}); + +test("custom gate prompt is used as the approval message", async () => { + const seen: string[] = []; + const { deps } = recorder(); + await executeFlow(flow({ skill: APPROVAL_GATE, prompt: "Ship to production?" }), "", { + ...deps, + requestApproval: async (msg) => { seen.push(msg); return true; }, + }); + assert.match(seen[0], /Approval Required: Ship to production\?/); +}); + +test("gate with no prompt falls back to a summary of progress so far", async () => { + const seen: string[] = []; + const { deps } = recorder(["THE-RESULT"]); + await executeFlow( + flow({ skill: "one", prompt: "do one" }, { skill: APPROVAL_GATE, prompt: "" }), + "", + { ...deps, requestApproval: async (msg) => { seen.push(msg); return true; } }, + ); + assert.match(seen[0], /paused at step 2\/2/); + assert.match(seen[0], /THE-RESULT/); +}); + +// ── Progress events ──────────────────────────────────────────────────── + +test("emits a start/step/done event stream", async () => { + const events: FlowEvent[] = []; + const { deps } = recorder(); + await executeFlow(flow({ skill: "one", prompt: "x" }, { skill: "two", prompt: "y" }), "", { + ...deps, + onProgress: (e) => events.push(e), + }); + + assert.deepEqual(events.map((e) => e.type), [ + "flow_start", "step_start", "step_done", "step_start", "step_done", "flow_done", + ]); + assert.deepEqual(events[0], { type: "flow_start", flow: "test-flow", totalSteps: 2 }); +}); + +test("emits approval and abort events when a gate denies", async () => { + const events: FlowEvent[] = []; + const { deps } = recorder(); + await executeFlow(flow({ skill: APPROVAL_GATE, prompt: "?" }), "", { + ...deps, + requestApproval: async () => false, + onProgress: (e) => events.push(e), + }); + + assert.deepEqual(events.map((e) => e.type), [ + "flow_start", "approval_requested", "approval_resolved", "flow_aborted", + ]); +}); + +// ── Error handling ───────────────────────────────────────────────────── + +test("a failing step propagates to the caller", async () => { + await assert.rejects( + executeFlow(flow({ skill: "boom", prompt: "x" }), "", { + runStep: async () => { throw new Error("model exploded"); }, + }), + /model exploded/, + ); +}); + +test("an empty flow completes without running anything", async () => { + const { prompts, deps } = recorder(); + const r = await executeFlow(flow(), "", deps); + assert.equal(r.completed, true); + assert.equal(r.steps.length, 0); + assert.equal(prompts.length, 0); +}); diff --git a/test/run-flow.test.ts b/test/run-flow.test.ts new file mode 100644 index 0000000..c073117 --- /dev/null +++ b/test/run-flow.test.ts @@ -0,0 +1,144 @@ +// Tests for the runFlow convenience wrapper. These deliberately use flows whose +// only step is an approval gate — a gate is resolved before any agent call, so +// the whole wrapper (flow lookup, approval policy, result shape) is exercised +// without needing an API key or spending a token. + +import test from "node:test"; +import assert from "node:assert/strict"; +import { mkdtemp, rm, mkdir, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +// Imported from dist/ (not src/) because run-flow.ts has runtime imports of +// sibling modules using ".js" specifiers, which don't resolve under +// --experimental-strip-types. Same convention as test/mcp.test.ts. +import { runFlow } from "../dist/run-flow.js"; + +async function agentWith(files: Record): Promise { + const dir = await mkdtemp(join(tmpdir(), "gitagent-runflow-")); + await mkdir(join(dir, "workflows"), { recursive: true }); + for (const [name, body] of Object.entries(files)) { + await writeFile(join(dir, "workflows", name), body, "utf-8"); + } + return dir; +} + +const GATE_ONLY = `name: gate-only +description: A single approval gate, nothing else +steps: + - skill: __approval_gate__ + prompt: "Continue?" + channel: telegram +`; + +// ── Flow lookup ──────────────────────────────────────────────────────── + +test("unknown flow name errors and lists what is available", async () => { + const dir = await agentWith({ "gate-only.yaml": GATE_ONLY }); + try { + await assert.rejects( + runFlow({ agentDir: dir, flow: "does-not-exist" }), + /Flow "does-not-exist" not found.*available: gate-only/s, + ); + } finally { + await rm(dir, { recursive: true, force: true }); + } +}); + +test("a markdown reference workflow is rejected as not runnable", async () => { + const dir = await agentWith({ + "notes.md": "---\nname: notes\ndescription: just a reference doc\n---\n\nSome prose.\n", + }); + try { + await assert.rejects( + runFlow({ agentDir: dir, flow: "notes" }), + /reference workflow.*not a runnable SkillFlow/s, + ); + } finally { + await rm(dir, { recursive: true, force: true }); + } +}); + +// ── Approval policy ──────────────────────────────────────────────────── + +test("gates deny by default — no policy given", async () => { + const dir = await agentWith({ "gate-only.yaml": GATE_ONLY }); + try { + const r = await runFlow({ agentDir: dir, flow: "gate-only" }); + assert.equal(r.completed, false); + assert.match(r.abortReason!, /no approval handler available/); + assert.deepEqual(r.usage, []); + } finally { + await rm(dir, { recursive: true, force: true }); + } +}); + +test('approve: "deny" is explicit but behaves the same', async () => { + const dir = await agentWith({ "gate-only.yaml": GATE_ONLY }); + try { + const r = await runFlow({ agentDir: dir, flow: "gate-only", approve: "deny" }); + assert.equal(r.completed, false); + } finally { + await rm(dir, { recursive: true, force: true }); + } +}); + +test('approve: "auto" lets gates through', async () => { + const dir = await agentWith({ "gate-only.yaml": GATE_ONLY }); + try { + const r = await runFlow({ agentDir: dir, flow: "gate-only", approve: "auto" }); + assert.equal(r.completed, true); + assert.equal(r.steps.length, 0); // a gate produces no step result + } finally { + await rm(dir, { recursive: true, force: true }); + } +}); + +test("a custom approver receives the message and the step's channel", async () => { + const dir = await agentWith({ "gate-only.yaml": GATE_ONLY }); + const seen: { message: string; channel?: string }[] = []; + try { + const r = await runFlow({ + agentDir: dir, + flow: "gate-only", + approve: async (message, channel) => { seen.push({ message, channel }); return true; }, + }); + assert.equal(r.completed, true); + assert.equal(seen.length, 1); + assert.equal(seen[0].channel, "telegram"); + assert.match(seen[0].message, /Approval Required: Continue\?/); + } finally { + await rm(dir, { recursive: true, force: true }); + } +}); + +// ── Result shape ─────────────────────────────────────────────────────── + +test("result carries usage totals even when nothing ran", async () => { + const dir = await agentWith({ "gate-only.yaml": GATE_ONLY }); + try { + const r = await runFlow({ agentDir: dir, flow: "gate-only", approve: "auto" }); + assert.deepEqual(r.usage, []); + assert.equal(r.totalCostUsd, 0); + assert.equal(r.totalTokens, 0); + assert.equal(r.flow, "gate-only"); + } finally { + await rm(dir, { recursive: true, force: true }); + } +}); + +test("progress events are forwarded through the wrapper", async () => { + const dir = await agentWith({ "gate-only.yaml": GATE_ONLY }); + const events: string[] = []; + try { + await runFlow({ + agentDir: dir, + flow: "gate-only", + approve: "auto", + onProgress: (e) => events.push(e.type), + }); + assert.deepEqual(events, ["flow_start", "approval_requested", "approval_resolved", "flow_done"]); + } finally { + await rm(dir, { recursive: true, force: true }); + } +}); From 56ac4316a8e366493a983fc56a581a3b5f3711ae Mon Sep 17 00:00:00 2001 From: Nivesh353 Date: Tue, 28 Jul 2026 13:20:50 +0530 Subject: [PATCH 2/2] fix: tighten @flow matching and flow name resolution MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-ups on #89. The @flow trigger required only a word boundary after the name, so "@daily-report.com" fired the flow with input ".com" — a "." counts as a boundary. Require whitespace or end-of-line instead, which rejects it outright. The same line existed in the voice server; fixed there too. resolveFlow matched by name regardless of type, then errored if the result turned out to be a reference workflow. With both notes.md and notes.yaml present, readdir order decided which one won, so a runnable flow could be reported as "not runnable". Prefer a runnable match and fall back to any match, keeping the more useful error when only a markdown workflow exists. Not changed: step 0 passing the user's input both through {input} and the context block. Emptying the context block drops the input entirely for flows whose first step doesn't template {input}, which is most of them. --- src/index.ts | 6 ++++-- src/run-flow.ts | 8 +++++++- test/run-flow.test.ts | 15 +++++++++++++++ 3 files changed, 26 insertions(+), 3 deletions(-) diff --git a/src/index.ts b/src/index.ts index e01d40e..78a33f0 100644 --- a/src/index.ts +++ b/src/index.ts @@ -862,8 +862,10 @@ async function main(): Promise { } // 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. - const flowMatch = trimmed.match(/^@([a-z0-9]+(?:-[a-z0-9]+)*)\b\s*([\s\S]*)$/); + // 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()); diff --git a/src/run-flow.ts b/src/run-flow.ts index d3fba97..d9699d8 100644 --- a/src/run-flow.ts +++ b/src/run-flow.ts @@ -80,7 +80,13 @@ async function resolveFlow(agentDir: string, flow: string | SkillFlowDefinition) if (typeof flow !== "string") return flow; const workflows = await discoverWorkflows(agentDir); - const meta = workflows.find((w) => w.name === flow); + // Prefer a runnable flow: a name can collide across formats (notes.md and + // notes.yaml both discover as "notes"), and readdir order is filesystem + // dependent, so an unfiltered find could return the markdown one and report + // "not runnable" for a flow that exists. The second lookup is only so the + // error below can say "that's a reference workflow" instead of "not found". + const meta = workflows.find((w) => w.name === flow && w.type === "flow") + ?? workflows.find((w) => w.name === flow); if (!meta) { const runnable = workflows.filter((w) => w.type === "flow").map((w) => w.name); diff --git a/test/run-flow.test.ts b/test/run-flow.test.ts index c073117..7bef8ab 100644 --- a/test/run-flow.test.ts +++ b/test/run-flow.test.ts @@ -45,6 +45,21 @@ test("unknown flow name errors and lists what is available", async () => { } }); +test("a runnable flow wins over a same-named reference workflow", async () => { + // Both discover as "gate-only". readdir order is filesystem-dependent, so + // resolution must prefer the runnable one rather than whichever came first. + const dir = await agentWith({ + "gate-only.md": "---\nname: gate-only\ndescription: a doc that shadows the flow\n---\n\nProse.\n", + "gate-only.yaml": GATE_ONLY, + }); + try { + const r = await runFlow({ agentDir: dir, flow: "gate-only", approve: "auto" }); + assert.equal(r.completed, true); // resolved the YAML, not the markdown + } finally { + await rm(dir, { recursive: true, force: true }); + } +}); + test("a markdown reference workflow is rejected as not runnable", async () => { const dir = await agentWith({ "notes.md": "---\nname: notes\ndescription: just a reference doc\n---\n\nSome prose.\n",