diff --git a/package.json b/package.json index c8cf7b5..a335ae0 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@clawnify/clawflow", - "version": "1.5.0", + "version": "1.5.1", "description": "The n8n for agents. A declarative, AI-native workflow format that agents can read, write, and run.", "type": "module", "main": "./dist/index.js", diff --git a/skills/clawflow/SKILL.md b/skills/clawflow/SKILL.md index 9e92949..972a3f6 100644 --- a/skills/clawflow/SKILL.md +++ b/skills/clawflow/SKILL.md @@ -566,6 +566,10 @@ flows/ **Rules:** - `flow_run` uses the latest published version by default. Falls back to draft if no versions exist. +- Incoming triggers (`POST /flows/:name/run` on the flow server — webhooks, HTTP + triggers, dashboard runs) resolve the same way: latest published version, draft + only when nothing is published. So a draft edit does not change what a live + trigger executes once the flow has been published at least once. - `flow_run file: "my-flow" draft: true` — explicitly run the working copy - `flow_run file: "my-flow" version: 2` — run a specific version - `flow_read file: "my-flow" version: 1` — inspect a specific published version diff --git a/src/core/serve.ts b/src/core/serve.ts index 917788c..a86aa83 100644 --- a/src/core/serve.ts +++ b/src/core/serve.ts @@ -4,6 +4,7 @@ import * as path from "path"; import type { FlowDefinition, ServeConfig } from "./types.js"; import type { FlowRunner } from "./runner.js"; import { validateFlow } from "./validate.js"; +import { readLatestVersion } from "./manage.js"; // ---- Flow Server ---------------------------------------------------------------- // Lightweight HTTP server that runs flows on POST. Trigger semantics (webhooks, @@ -18,6 +19,11 @@ import { validateFlow } from "./validate.js"; export interface FlowServerOpts { runner: FlowRunner; serve: ServeConfig; + /** + * Workspace root — where published versions live (.clawflow/versions). + * Defaults to $OPENCLAW_WORKSPACE, then cwd, matching resolveFlowsDir. + */ + workspace?: string; logger?: { info: (msg: string) => void; warn: (msg: string) => void; @@ -31,23 +37,42 @@ const MAX_BODY_BYTES = 1_048_576; // 1 MB // (OpenClaw calls it during discovery and again at gateway startup). let activeServer: http.Server | null = null; -function resolveFlowsDir(serve: ServeConfig): string { - return ( - serve.flowsDir ?? - path.join(process.env.OPENCLAW_WORKSPACE ?? process.cwd(), "flows") - ); +function resolveWorkspace(workspace?: string): string { + return workspace ?? process.env.OPENCLAW_WORKSPACE ?? process.cwd(); } +function resolveFlowsDir(serve: ServeConfig, workspace: string): string { + return serve.flowsDir ?? path.join(workspace, "flows"); +} + +/** + * Resolve what an incoming trigger should execute: the latest PUBLISHED + * version when the flow has one, else the draft. + * + * Same precedence as the flow_run tool — a webhook and an agent run must never + * execute different definitions of the same flow, or "publish" means nothing + * for every off-box caller (webhooks, HTTP triggers, the dashboard). + * Unpublished flows still run from the draft so a flow works the moment it's + * written. + */ function loadFlow( + workspace: string, flowsDir: string, flowName: string, -): FlowDefinition | null { +): { def: FlowDefinition; source: string } | null { const safe = flowName.replace(/[^a-zA-Z0-9_-]/g, ""); if (!safe) return null; + + const latest = readLatestVersion(workspace, safe); + if (latest) return { def: latest.def, source: `v${latest.version}` }; + const file = path.join(flowsDir, `${safe}.json`); if (!fs.existsSync(file)) return null; try { - return JSON.parse(fs.readFileSync(file, "utf8")) as FlowDefinition; + return { + def: JSON.parse(fs.readFileSync(file, "utf8")) as FlowDefinition, + source: "draft (no published versions)", + }; } catch { return null; } @@ -89,7 +114,8 @@ export function startFlowServer(opts: FlowServerOpts): http.Server { const { runner, serve, logger } = opts; const basePath = (serve.path ?? "/flows").replace(/\/+$/, ""); - const flowsDir = resolveFlowsDir(serve); + const workspace = resolveWorkspace(opts.workspace); + const flowsDir = resolveFlowsDir(serve, workspace); const log = logger ?? { info: console.log, warn: console.warn, @@ -150,11 +176,12 @@ export function startFlowServer(opts: FlowServerOpts): http.Server { const flowName = match[1]; try { - const flowDef = loadFlow(flowsDir, flowName); - if (!flowDef) { + const loaded = loadFlow(workspace, flowsDir, flowName); + if (!loaded) { json(res, 404, { error: `Flow not found: ${flowName}` }); return; } + const flowDef = loaded.def; // Parse request body — entire body becomes the flow's inputs payload. let inputs: unknown = {}; @@ -170,7 +197,7 @@ export function startFlowServer(opts: FlowServerOpts): http.Server { // Fire-and-forget: start the flow, return immediately with instanceId const instanceId = crypto.randomUUID(); - log.info(`[clawflow] run → ${flowName} (${instanceId})`); + log.info(`[clawflow] run → ${flowName} ${loaded.source} (${instanceId})`); // Respond 202 before the flow runs json(res, 202, { ok: true, instanceId, flow: flowName }); diff --git a/src/plugin/index.ts b/src/plugin/index.ts index 2c0fa22..ff0a34e 100644 --- a/src/plugin/index.ts +++ b/src/plugin/index.ts @@ -103,6 +103,9 @@ function register(api: PluginApi) { startFlowServer({ runner, serve: pluginCfg.serve, + // Same workspace the flow_* tools use, so a triggered run resolves the + // published version exactly like flow_run does. + workspace, logger: api.logger, }); } diff --git a/tests/manage.test.ts b/tests/manage.test.ts index d77e820..8e35c84 100644 --- a/tests/manage.test.ts +++ b/tests/manage.test.ts @@ -119,9 +119,9 @@ describe("publishDraft", () => { }); }); -// ---- POST /flows/validate (flow server) ------------------------------------------- +// ---- flow server routes (validate + run resolution) -------------------------------- -describe("flow server validate route", () => { +describe("flow server routes", () => { let server: ReturnType; let base: string; @@ -133,6 +133,7 @@ describe("flow server validate route", () => { server = startFlowServer({ runner, serve: { port: 0, path: "/flows" }, + workspace, logger: { info: () => {}, warn: () => {}, error: () => {} }, }); await new Promise((resolve) => server.on("listening", resolve)); @@ -210,6 +211,38 @@ describe("flow server validate route", () => { assert.equal(bad2.status, 400); }); + // The run route must resolve the same definition flow_run would: published + // version first, draft only when nothing is published. A corrupted draft is + // the discriminator — if the server still answers 202, it never read it. + it("runs the latest published version, not the draft", async () => { + writeDraft("srv-published"); + publishDraft(workspace, "srv-published"); + fs.writeFileSync( + path.join(workspace, "flows", "srv-published.json"), + "{ not json at all", + ); + + const res = await fetch(`${base}/srv-published/run`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: "{}", + }); + assert.equal(res.status, 202); + const body = (await res.json()) as { ok: boolean; flow: string }; + assert.equal(body.ok, true); + assert.equal(body.flow, "srv-published"); + }); + + it("falls back to the draft when nothing is published", async () => { + writeDraft("srv-draft-only"); + const res = await fetch(`${base}/srv-draft-only/run`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: "{}", + }); + assert.equal(res.status, 202); + }); + it("does not shadow the run route", async () => { // A flow literally named "validate" must still be runnable by name. const res = await fetch(`${base}/validate/run`, { method: "POST", body: "{}" });