From 7187e4c1f3c30f63a25c15a416978bc9e4ae2376 Mon Sep 17 00:00:00 2001 From: pallaoro Date: Mon, 3 Aug 2026 11:29:41 +0200 Subject: [PATCH] feat: POST /flows/validate route + core/manage module (publishDraft), bump to 1.5.0 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit serve.ts gains a pure validation endpoint that runs validateFlow against the live in-process step registry — the one place off-box callers (Clawnify hook server) can get registry-correct validation for flows using plugin-registered custom steps (clawnify_app, clawnify_action, ...). Draft/version semantics (resolveFlowFile, versionsDir, listVersions, readVersion, readLatestVersion, publishDraft) move from plugin closures into src/core/manage.ts so out-of-process callers import the same implementation instead of mirroring it. flow_publish keeps its in-process validation and now delegates the snapshot to publishDraft — no behavior change. --- package.json | 2 +- src/core/manage.ts | 121 ++++++++++++++++++++++++ src/core/serve.ts | 32 +++++++ src/index.ts | 9 ++ src/plugin/index.ts | 62 ++++-------- tests/manage.test.ts | 220 +++++++++++++++++++++++++++++++++++++++++++ 6 files changed, 402 insertions(+), 44 deletions(-) create mode 100644 src/core/manage.ts create mode 100644 tests/manage.test.ts diff --git a/package.json b/package.json index 7aab6a0..c8cf7b5 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@clawnify/clawflow", - "version": "1.4.1", + "version": "1.5.0", "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/src/core/manage.ts b/src/core/manage.ts new file mode 100644 index 0000000..e93c7f2 --- /dev/null +++ b/src/core/manage.ts @@ -0,0 +1,121 @@ +import * as fs from "fs"; +import * as path from "path"; +import type { FlowDefinition } from "./types.js"; + +// ---- Draft / Version Management ------------------------------------------------- +// Engine-owned draft + published-version semantics: the draft file convention +// (workspace/flows/.json), the versions directory layout +// (.clawflow/versions//.json), next-version assignment, and the +// version stamp. +// +// Used by the plugin's flow_* tools in-process, and by out-of-process platform +// callers (e.g. the Clawnify hook server) that import these functions directly +// from dist — so the layout exists in exactly one place. +// +// NOTE: none of these functions validate. Validation needs the caller's step +// registry (custom steps registered by sibling plugins live in the gateway +// process), so each caller validates first with the registry it has: the +// plugin tools use the in-process defaultRegistry, the hook server uses the +// flow server's POST /flows/validate route. + +/** Resolve a file param to an absolute path using workspace conventions. */ +export function resolveFlowFile(workspace: string, file: string): string { + if (file.startsWith("/")) return file; + if (file.includes("/")) return path.join(workspace, file); + const name = file.replace(/\.json$/, ""); + return path.join(workspace, "flows", `${name}.json`); +} + +/** Get the versions directory for a flow name. */ +export function versionsDir(workspace: string, flowName: string): string { + return path.join(workspace, ".clawflow", "versions", flowName); +} + +/** List all published version numbers for a flow, sorted ascending. */ +export function listVersions(workspace: string, flowName: string): number[] { + const dir = versionsDir(workspace, flowName); + if (!fs.existsSync(dir)) return []; + return fs.readdirSync(dir) + .filter((f: string) => /^\d+\.json$/.test(f)) + .map((f: string) => parseInt(f, 10)) + .sort((a: number, b: number) => a - b); +} + +/** Read a specific published version. Returns null if not found. */ +export function readVersion( + workspace: string, + flowName: string, + version: number, +): FlowDefinition | null { + const file = path.join(versionsDir(workspace, flowName), `${version}.json`); + if (!fs.existsSync(file)) return null; + return JSON.parse(fs.readFileSync(file, "utf-8")) as FlowDefinition; +} + +/** Get the latest published version definition. Returns null if none published. */ +export function readLatestVersion( + workspace: string, + flowName: string, +): { version: number; def: FlowDefinition } | null { + const versions = listVersions(workspace, flowName); + if (versions.length === 0) return null; + const latest = versions[versions.length - 1]; + const def = readVersion(workspace, flowName, latest); + if (!def) return null; + return { version: latest, def }; +} + +export interface PublishResult { + flow: string; + version: number; + file: string; + totalVersions: number; +} + +/** + * Publish the current draft of a flow as a new numbered version. + * + * Reads the draft from `file` (workspace conventions, see resolveFlowFile), + * assigns the next version number (auto-incrementing integer), stamps it into + * the definition, and saves an immutable copy to + * .clawflow/versions//.json. After publishing, flow_run uses this + * version by default. + * + * Throws Error("Draft not found: …") when the draft file is missing and + * Error("Failed to parse …") when it isn't valid JSON. Does NOT validate the + * definition — callers validate first (see module note). + */ +export function publishDraft(workspace: string, file: string): PublishResult { + const abs = resolveFlowFile(workspace, file); + if (!fs.existsSync(abs)) { + throw new Error(`Draft not found: ${abs}`); + } + + let flowDef: FlowDefinition; + try { + flowDef = JSON.parse(fs.readFileSync(abs, "utf-8")) as FlowDefinition; + } catch (err) { + throw new Error( + `Failed to parse ${abs}: ${err instanceof Error ? err.message : String(err)}`, + ); + } + + const flowName = path.basename(abs, ".json"); + const versions = listVersions(workspace, flowName); + const nextVersion = versions.length > 0 ? versions[versions.length - 1] + 1 : 1; + + // Stamp the version number into the definition + flowDef.version = String(nextVersion); + + const dir = versionsDir(workspace, flowName); + fs.mkdirSync(dir, { recursive: true }); + const versionFile = path.join(dir, `${nextVersion}.json`); + fs.writeFileSync(versionFile, JSON.stringify(flowDef, null, 2) + "\n"); + + return { + flow: flowDef.flow, + version: nextVersion, + file: versionFile, + totalVersions: nextVersion, + }; +} diff --git a/src/core/serve.ts b/src/core/serve.ts index 7600134..917788c 100644 --- a/src/core/serve.ts +++ b/src/core/serve.ts @@ -3,6 +3,7 @@ import * as fs from "fs"; import * as path from "path"; import type { FlowDefinition, ServeConfig } from "./types.js"; import type { FlowRunner } from "./runner.js"; +import { validateFlow } from "./validate.js"; // ---- Flow Server ---------------------------------------------------------------- // Lightweight HTTP server that runs flows on POST. Trigger semantics (webhooks, @@ -11,6 +12,7 @@ import type { FlowRunner } from "./runner.js"; // // Endpoints: // POST /:basePath/:flowName/run — run a flow with the JSON body as inputs +// POST /:basePath/validate — statically validate a flow definition // GET /:basePath/health — health check export interface FlowServerOpts { @@ -104,6 +106,36 @@ export function startFlowServer(opts: FlowServerOpts): http.Server { return; } + // Validate a flow definition: POST /:basePath/validate + // Pure static validation against THIS process's live step registry — + // custom steps registered by sibling plugins (e.g. Clawnify's + // clawnify_app/clawnify_action) only exist in the gateway process, so this + // is the one place an off-box caller can get registry-correct validation. + // No state, no execution; safe to leave unauthenticated like /health. + if (req.method === "POST" && pathname === `${basePath}/validate`) { + try { + const rawBody = await readBody(req); + let def: unknown; + try { + def = rawBody ? JSON.parse(rawBody) : null; + } catch { + json(res, 400, { error: "Invalid JSON body" }); + return; + } + if (!def || typeof def !== "object" || Array.isArray(def)) { + json(res, 400, { error: "Body must be a flow definition object" }); + return; + } + json(res, 200, validateFlow(def as FlowDefinition)); + } catch (err) { + log.error( + `[clawflow] validate error: ${err instanceof Error ? err.message : String(err)}`, + ); + json(res, 500, { error: "Internal server error" }); + } + return; + } + // Run a flow: POST /:basePath/:flowName/run const runPattern = new RegExp( `^${basePath.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")}/([a-zA-Z0-9_-]+)/run$`, diff --git a/src/index.ts b/src/index.ts index e685d54..5a19428 100644 --- a/src/index.ts +++ b/src/index.ts @@ -58,3 +58,12 @@ export { parseDuration, MODEL_MAP, DEFAULT_MODEL, NODE_KEYS } from "./core/types export { startFlowServer } from "./core/serve.js"; export type { FlowServerOpts } from "./core/serve.js"; export type { ServeConfig } from "./core/types.js"; +export { + publishDraft, + resolveFlowFile, + versionsDir, + listVersions, + readVersion, + readLatestVersion, +} from "./core/manage.js"; +export type { PublishResult } from "./core/manage.js"; diff --git a/src/plugin/index.ts b/src/plugin/index.ts index e206330..2c0fa22 100644 --- a/src/plugin/index.ts +++ b/src/plugin/index.ts @@ -5,6 +5,13 @@ import { FlowRunner, sendEvent } from "../core/runner.js"; import { startFlowServer } from "../core/serve.js"; import { validateFlow } from "../core/validate.js"; +import { + publishDraft, + listVersions as listVersionsCore, + readLatestVersion as readLatestVersionCore, + readVersion as readVersionCore, + resolveFlowFile as resolveFlowFileCore, +} from "../core/manage.js"; import type { FlowDefinition, FlowNode, PluginConfig, BranchNode, ConditionNode, LoopNode, ParallelNode } from "../core/types.js"; // ---- OpenClaw Plugin: clawflow --------------------------------------------------- @@ -185,47 +192,27 @@ function register(api: PluginApi) { } // ---- Shared helpers ------------------------------------------------------------ + // Draft/version semantics live in core/manage.ts (engine-owned, shared with + // out-of-process callers); these closures just bind the plugin's workspace. /** Resolve a file param to an absolute path using workspace conventions. */ function resolveFlowFile(file: string): string { - const base = workspace; - if (file.startsWith("/")) return file; - if (file.includes("/")) return path.join(base, file); - const name = file.replace(/\.json$/, ""); - return path.join(base, "flows", `${name}.json`); - } - - /** Get the versions directory for a flow name. */ - function versionsDir(flowName: string): string { - const base = workspace; - return path.join(base, ".clawflow", "versions", flowName); + return resolveFlowFileCore(workspace, file); } /** List all published version numbers for a flow, sorted ascending. */ function listVersions(flowName: string): number[] { - const dir = versionsDir(flowName); - if (!fs.existsSync(dir)) return []; - return fs.readdirSync(dir) - .filter((f: string) => /^\d+\.json$/.test(f)) - .map((f: string) => parseInt(f, 10)) - .sort((a: number, b: number) => a - b); + return listVersionsCore(workspace, flowName); } /** Read a specific published version. Returns null if not found. */ function readVersion(flowName: string, version: number): FlowDefinition | null { - const file = path.join(versionsDir(flowName), `${version}.json`); - if (!fs.existsSync(file)) return null; - return JSON.parse(fs.readFileSync(file, "utf-8")) as FlowDefinition; + return readVersionCore(workspace, flowName, version); } /** Get the latest published version definition. Returns null if none published. */ function readLatestVersion(flowName: string): { version: number; def: FlowDefinition } | null { - const versions = listVersions(flowName); - if (versions.length === 0) return null; - const latest = versions[versions.length - 1]; - const def = readVersion(flowName, latest); - if (!def) return null; - return { version: latest, def }; + return readLatestVersionCore(workspace, flowName); } // ---- flow_create -------------------------------------------------------------- @@ -1262,7 +1249,6 @@ modify the draft without affecting published versions.`, params: { file: string }, ) { const fs = await import("fs"); - const pathMod = await import("path"); const abs = resolveFlowFile(params.file); if (!fs.existsSync(abs)) { @@ -1297,30 +1283,20 @@ modify the draft without affecting published versions.`, }; } - const flowName = pathMod.basename(abs, ".json"); - const versions = listVersions(flowName); - const nextVersion = versions.length > 0 ? versions[versions.length - 1] + 1 : 1; - - // Stamp the version number into the definition - flowDef.version = String(nextVersion); - - const dir = versionsDir(flowName); - fs.mkdirSync(dir, { recursive: true }); - const versionFile = pathMod.join(dir, `${nextVersion}.json`); - fs.writeFileSync(versionFile, JSON.stringify(flowDef, null, 2) + "\n"); + const published = publishDraft(workspace, params.file); return { content: [ { type: "text", - text: `Published "${flowDef.flow}" as v${nextVersion}. flow_run will now use this version by default.\nFile: ${versionFile}`, + text: `Published "${published.flow}" as v${published.version}. flow_run will now use this version by default.\nFile: ${published.file}`, }, ], details: { - flow: flowDef.flow, - version: nextVersion, - file: versionFile, - totalVersions: nextVersion, + flow: published.flow, + version: published.version, + file: published.file, + totalVersions: published.totalVersions, }, }; }, diff --git a/tests/manage.test.ts b/tests/manage.test.ts new file mode 100644 index 0000000..d77e820 --- /dev/null +++ b/tests/manage.test.ts @@ -0,0 +1,220 @@ +import { describe, it, before, after } from "node:test"; +import assert from "node:assert/strict"; +import * as fs from "fs"; +import * as path from "path"; +import * as os from "os"; + +import { + FlowRunner, + defaultRegistry, + listVersions, + publishDraft, + readLatestVersion, + readVersion, + resolveFlowFile, + startFlowServer, +} from "../src/index.js"; +import type { FlowDefinition } from "../src/index.js"; + +const tmpDir = path.join(os.tmpdir(), `ocf-manage-test-${Date.now()}`); +const workspace = path.join(tmpDir, "workspace"); + +function cleanup() { + fs.rmSync(tmpDir, { recursive: true, force: true }); +} + +const simpleFlow: FlowDefinition = { + flow: "manage-test", + nodes: [{ name: "step1", do: "code", run: "return 1", output: "result" }], +}; + +function writeDraft(name: string, def: unknown = simpleFlow): string { + const file = path.join(workspace, "flows", `${name}.json`); + fs.mkdirSync(path.dirname(file), { recursive: true }); + fs.writeFileSync(file, JSON.stringify(def, null, 2)); + return file; +} + +// ---- resolveFlowFile ------------------------------------------------------------ + +describe("resolveFlowFile", () => { + after(cleanup); + + it("resolves plain names to workspace/flows/.json", () => { + assert.equal( + resolveFlowFile(workspace, "my-flow"), + path.join(workspace, "flows", "my-flow.json"), + ); + }); + + it("strips a .json suffix from plain names", () => { + assert.equal( + resolveFlowFile(workspace, "my-flow.json"), + path.join(workspace, "flows", "my-flow.json"), + ); + }); + + it("resolves relative paths against the workspace", () => { + assert.equal( + resolveFlowFile(workspace, "custom/dir/f.json"), + path.join(workspace, "custom/dir/f.json"), + ); + }); + + it("passes absolute paths through", () => { + assert.equal(resolveFlowFile(workspace, "/tmp/x.json"), "/tmp/x.json"); + }); +}); + +// ---- publishDraft + version readers ---------------------------------------------- + +describe("publishDraft", () => { + after(cleanup); + + it("publishes v1 stamped into the definition, then v2", () => { + writeDraft("incr"); + const first = publishDraft(workspace, "incr"); + assert.equal(first.version, 1); + assert.equal(first.flow, "manage-test"); + assert.equal(first.totalVersions, 1); + const onDisk = JSON.parse(fs.readFileSync(first.file, "utf-8")); + assert.equal(onDisk.version, "1"); + + const second = publishDraft(workspace, "incr"); + assert.equal(second.version, 2); + assert.deepEqual(listVersions(workspace, "incr"), [1, 2]); + }); + + it("does not modify the draft file", () => { + const draftPath = writeDraft("immutable"); + const before = fs.readFileSync(draftPath, "utf-8"); + publishDraft(workspace, "immutable"); + assert.equal(fs.readFileSync(draftPath, "utf-8"), before); + }); + + it("round-trips through readVersion / readLatestVersion", () => { + writeDraft("readback"); + publishDraft(workspace, "readback"); + publishDraft(workspace, "readback"); + + const v1 = readVersion(workspace, "readback", 1); + assert.equal(v1?.version, "1"); + assert.equal(readVersion(workspace, "readback", 99), null); + + const latest = readLatestVersion(workspace, "readback"); + assert.equal(latest?.version, 2); + assert.equal(latest?.def.version, "2"); + assert.equal(readLatestVersion(workspace, "never-published"), null); + }); + + it("throws when the draft is missing", () => { + assert.throws(() => publishDraft(workspace, "nope"), /Draft not found/); + }); + + it("throws when the draft is not valid JSON", () => { + const file = path.join(workspace, "flows", "broken.json"); + fs.mkdirSync(path.dirname(file), { recursive: true }); + fs.writeFileSync(file, "{ not json"); + assert.throws(() => publishDraft(workspace, "broken"), /Failed to parse/); + }); +}); + +// ---- POST /flows/validate (flow server) ------------------------------------------- + +describe("flow server validate route", () => { + let server: ReturnType; + let base: string; + + before(async () => { + const runner = new FlowRunner({ + stateDir: path.join(tmpDir, "state"), + memoryDir: path.join(tmpDir, "memory"), + }); + server = startFlowServer({ + runner, + serve: { port: 0, path: "/flows" }, + logger: { info: () => {}, warn: () => {}, error: () => {} }, + }); + await new Promise((resolve) => server.on("listening", resolve)); + const addr = server.address(); + if (!addr || typeof addr === "string") throw new Error("no address"); + base = `http://127.0.0.1:${addr.port}/flows`; + }); + + after(() => { + server.close(); + cleanup(); + }); + + it("validates a good definition", async () => { + const res = await fetch(`${base}/validate`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(simpleFlow), + }); + assert.equal(res.status, 200); + const body = (await res.json()) as { ok: boolean; errors: unknown[] }; + assert.equal(body.ok, true); + assert.deepEqual(body.errors, []); + }); + + it("reports node-level errors for a bad definition", async () => { + const res = await fetch(`${base}/validate`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ flow: "bad", nodes: [{ name: "x", do: "nope" }] }), + }); + assert.equal(res.status, 200); + const body = (await res.json()) as { + ok: boolean; + errors: { node?: string; message: string }[]; + }; + assert.equal(body.ok, false); + assert.ok(body.errors.some((e) => e.message.includes('Unknown node type "nope"'))); + }); + + it("knows custom steps registered in this process's registry", async () => { + defaultRegistry.register({ + name: "manage_test_step", + allowedKeys: ["message"], + run: () => ({}), + }); + const flow: FlowDefinition = { + flow: "custom", + nodes: [ + { + name: "c", + do: "manage_test_step" as unknown as "code", + // @ts-expect-error custom field not in built-in types + message: "hi", + } as unknown as FlowDefinition["nodes"][number], + ], + }; + const res = await fetch(`${base}/validate`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(flow), + }); + const body = (await res.json()) as { ok: boolean; errors: { message: string }[] }; + assert.equal(body.ok, true, JSON.stringify(body.errors)); + }); + + it("rejects malformed JSON and non-object bodies", async () => { + const bad1 = await fetch(`${base}/validate`, { method: "POST", body: "{ nope" }); + assert.equal(bad1.status, 400); + const bad2 = await fetch(`${base}/validate`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify([1, 2]), + }); + assert.equal(bad2.status, 400); + }); + + 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: "{}" }); + assert.equal(res.status, 404); // routed as run (no such flow saved), not validate + const body = (await res.json()) as { error: string }; + assert.match(body.error, /Flow not found: validate/); + }); +});