Registry-correct flow management for off-box callers, bump to 1.5.0

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",

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/<name>.json), the versions directory layout
+// (.clawflow/versions/<flowName>/<N>.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/<flowName>/<N>.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,
+  };
+}

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$`,

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";

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,
           },
         };
       },