From 0c228f38a83bceebea9aa6dcd3b0e306ffcbefb5 Mon Sep 17 00:00:00 2001 From: pallaoro Date: Fri, 24 Jul 2026 15:37:48 +0000 Subject: [PATCH] refactor(core): extract harness-neutral op layer + AgentInvoker seam + management API MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Preparation for multi-harness adapters (Hermes et al.) — no behavior change for the OpenClaw plugin path. - FlowOps (src/core/ops.ts): the eleven flow_* tool implementations move out of the OpenClaw plugin wrapper into a harness-neutral class returning plain { text, details? } results. - FLOW_OP_SPECS (src/core/op-specs.ts): tool names, descriptions, and JSON-schema parameters as data, so every harness registers the same surface from one source of truth. - Management API (src/core/manage.ts): optional loopback-only HTTP server (127.0.0.1, bearer token required, default port 18794) exposing the op surface for out-of-process harness shims: GET /ops (spec discovery), POST /ops/:name (execute), GET /ops/health. Started by the plugin only when a token is configured (manage.token or CLAWFLOW_ADMIN_TOKEN). - AgentInvoker (src/core/agent-invoker.ts): agent nodes now reach their runtime through an interface; OpenClawCliInvoker preserves the existing CLI behavior as the default. Other harnesses supply their own implementation via PluginConfig.agentInvoker. - The OpenClaw plugin (src/plugin/index.ts) shrinks to a thin adapter: spec-driven tool registration + the approval gate + server startup. - Tests: ops lifecycle, management server (auth, discovery, execute), custom invoker routing. 181 pass. --- README.md | 12 + package-lock.json | 4 +- package.json | 2 +- src/core/agent-invoker.ts | 103 +++ src/core/manage.ts | 163 ++++ src/core/op-specs.ts | 440 +++++++++ src/core/ops.ts | 1091 ++++++++++++++++++++++ src/core/runner.ts | 85 +- src/core/types.ts | 21 + src/index.ts | 9 +- src/plugin/index.ts | 1837 +------------------------------------ tests/core.test.ts | 42 +- tests/ops.test.ts | 266 ++++++ 13 files changed, 2197 insertions(+), 1878 deletions(-) create mode 100644 src/core/agent-invoker.ts create mode 100644 src/core/manage.ts create mode 100644 src/core/op-specs.ts create mode 100644 src/core/ops.ts create mode 100644 tests/ops.test.ts diff --git a/README.md b/README.md index a751129..c4ad9a1 100644 --- a/README.md +++ b/README.md @@ -563,6 +563,18 @@ Eleven tools registered in OpenClaw: } ``` +**Harness adapters:** the eleven tools are implemented harness-neutrally in `FlowOps` (`src/core/ops.ts`), with their names, descriptions, and JSON-schema parameters exported as data (`FLOW_OP_SPECS`). The OpenClaw plugin is a thin registrar over that surface; adapters for other agent runtimes register the same specs and forward calls — in-process via `FlowOps`, or out-of-process via the management API below. + +**Management API (optional):** a loopback-only HTTP server exposing the same op surface for out-of-process adapters. Binds `127.0.0.1` only and never starts without a bearer token (`manage.token` config or `CLAWFLOW_ADMIN_TOKEN` env; default port 18794). + +``` +GET /ops # op specs (name, description, parameters) — discovery +POST /ops/:name # execute an op; JSON body = params → { text, details? } +GET /ops/health # health check +``` + +**Agent invoker:** `agent` nodes reach their runtime through the `AgentInvoker` interface (`PluginConfig.agentInvoker`). The default is the OpenClaw CLI; embedders and other harnesses supply their own implementation. + --- ### 2. Standalone Node.js Runner (coming soon) diff --git a/package-lock.json b/package-lock.json index 27cfb74..d420137 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@clawnify/clawflow", - "version": "1.3.1", + "version": "1.5.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@clawnify/clawflow", - "version": "1.3.1", + "version": "1.5.0", "license": "MIT", "devDependencies": { "@types/node": "^25.5.0", 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/agent-invoker.ts b/src/core/agent-invoker.ts new file mode 100644 index 0000000..c64b393 --- /dev/null +++ b/src/core/agent-invoker.ts @@ -0,0 +1,103 @@ +// ---- Agent invoker -------------------------------------------------------------- +// The `agent` node delegates a task to a real agent runtime. Which runtime — and +// how it is reached — is a harness concern, not a flow concern, so the runner +// talks to this interface and the harness supplies the implementation. The +// default is the OpenClaw CLI; other harnesses (e.g. Hermes) provide their own +// via PluginConfig.agentInvoker. + +/** Selector for which agent/session receives the message. Field names follow + * the OpenClaw CLI selectors (the format's reference harness); other invokers + * map them to their runtime's nearest equivalent. */ +export interface AgentTarget { + /** Configured agent slug (e.g. "main"). */ + agent?: string; + /** Exact session key (e.g. "agent:main:slack:channel:agent"). */ + sessionKey?: string; + /** Explicit session id — the most specific selector. */ + sessionId?: string; + /** Delivery channel for the agent's reply. */ + channel?: string; +} + +export interface AgentInvokeOptions { + /** Hard timeout for the invocation, in milliseconds. */ + timeoutMs: number; + /** Flow-level env vars (state.env) to expose to the invocation. */ + env?: Record; +} + +export interface AgentInvoker { + /** Run one agent turn and return the agent's reply as text. */ + invoke( + message: string, + target: AgentTarget, + opts: AgentInvokeOptions, + ): Promise; +} + +// ---- Default implementation: OpenClaw CLI --------------------------------------- + +export class OpenClawCliInvoker implements AgentInvoker { + constructor(private opts: { defaultAgent?: string } = {}) {} + + // Build the `openclaw agent` selector args from the target. Mirrors + // OpenClaw's own selectors, which compose: + // --agent scopes to a configured agent + // --session-key an exact session key (agent::, or scoped to --agent) + // --session-id the most specific: one explicit session (scoped by --agent) + // --channel delivery channel for the reply + // We pass through whichever are set and let OpenClaw resolve/enforce consistency + // rather than re-asserting it here. When no target selector is set at all, fall + // back to a configured agent (defaultAgent > "main") so the call is routable. + private buildArgs(target: AgentTarget): string[] { + const { agent, sessionKey, sessionId, channel } = target; + const args: string[] = []; + if (agent) args.push("--agent", agent); + if (sessionKey) args.push("--session-key", sessionKey); + if (sessionId) args.push("--session-id", sessionId); + if (channel) args.push("--channel", channel); + if (!agent && !sessionKey && !sessionId) { + args.push("--agent", this.opts.defaultAgent ?? "main"); + } + return args; + } + + async invoke( + message: string, + target: AgentTarget, + opts: AgentInvokeOptions, + ): Promise { + const { execFile } = await import("child_process"); + const { promisify } = await import("util"); + const execFileAsync = promisify(execFile); + + // Check if openclaw CLI is available + try { + await execFileAsync("which", ["openclaw"]); + } catch { + throw new Error("openclaw CLI not found — agent nodes require the openclaw CLI to be installed"); + } + + const args = ["agent", ...this.buildArgs(target), "--message", message]; + + // Merge flow-level env vars into the child process environment. + // Set CLAWFLOW_NO_SERVE to prevent the child from binding the webhook port. + const env = { + ...process.env, + ...(opts.env ?? {}), + CLAWFLOW_NO_SERVE: "1", + }; + + try { + const { stdout } = await execFileAsync("openclaw", args, { + timeout: opts.timeoutMs, + maxBuffer: 10 * 1024 * 1024, // 10MB + env, + }); + return stdout.trim(); + } catch (err) { + const msg = err instanceof Error ? err.message : String(err); + throw new Error(`openclaw agent failed: ${msg}`); + } + } +} diff --git a/src/core/manage.ts b/src/core/manage.ts new file mode 100644 index 0000000..38f748a --- /dev/null +++ b/src/core/manage.ts @@ -0,0 +1,163 @@ +import * as http from "node:http"; +import { createHash, timingSafeEqual } from "node:crypto"; + +import type { FlowOps } from "./ops.js"; +import { FLOW_OP_SPECS } from "./op-specs.js"; + +// ---- Management API ------------------------------------------------------------- +// Loopback-only HTTP server exposing the flow_* operations, so out-of-process +// harness shims (e.g. a Hermes-side plugin) can register the same tool surface +// and forward calls without linking against this package. This is deliberately +// NOT the flow server (serve.ts): that one is tunnel-exposed for webhook +// triggers; this one binds 127.0.0.1 only and requires a bearer token. +// +// Endpoints: +// GET /ops — list op specs (name, description, parameters) for discovery +// POST /ops/:name — execute an op; JSON body = params; returns { text, details? } +// GET /ops/health — health check (auth required, like everything else) + +export interface ManageServerOpts { + ops: FlowOps; + /** Bearer token required on every request. */ + token: string; + /** Default: 18794. Use 0 for an ephemeral port (tests). */ + port?: number; + logger?: { + info: (msg: string) => void; + warn: (msg: string) => void; + error: (msg: string) => void; + }; +} + +export const DEFAULT_MANAGE_PORT = 18794; + +const MAX_BODY_BYTES = 1_048_576; // 1 MB + +// Guard against double-init when register() is called more than once +// (OpenClaw calls it during discovery and again at gateway startup). +let activeServer: http.Server | null = null; + +function json(res: http.ServerResponse, status: number, body: unknown): void { + const payload = JSON.stringify(body); + res.writeHead(status, { + "Content-Type": "application/json", + "Content-Length": Buffer.byteLength(payload), + }); + res.end(payload); +} + +function readBody(req: http.IncomingMessage): Promise { + return new Promise((resolve, reject) => { + const chunks: Buffer[] = []; + let size = 0; + req.on("data", (chunk: Buffer) => { + size += chunk.length; + if (size > MAX_BODY_BYTES) { + req.destroy(); + reject(new Error("Request body too large")); + return; + } + chunks.push(chunk); + }); + req.on("end", () => resolve(Buffer.concat(chunks).toString("utf8"))); + req.on("error", reject); + }); +} + +/** Constant-time bearer-token check (hash both sides so lengths always match). */ +function tokenMatches(header: string | undefined, token: string): boolean { + if (!header?.startsWith("Bearer ")) return false; + const presented = createHash("sha256").update(header.slice(7)).digest(); + const expected = createHash("sha256").update(token).digest(); + return timingSafeEqual(presented, expected); +} + +export function startManagementServer(opts: ManageServerOpts): http.Server { + if (activeServer) return activeServer; + + const { ops, token } = opts; + const port = opts.port ?? DEFAULT_MANAGE_PORT; + const log = opts.logger ?? { + info: console.log, + warn: console.warn, + error: console.error, + }; + + const opNames = new Set(FLOW_OP_SPECS.map((s) => s.name)); + + const server = http.createServer(async (req, res) => { + if (!tokenMatches(req.headers.authorization, token)) { + json(res, 401, { error: "Unauthorized" }); + return; + } + + const url = new URL(req.url ?? "/", "http://127.0.0.1"); + const pathname = url.pathname.replace(/\/+$/, "") || "/"; + + if (req.method === "GET" && pathname === "/ops/health") { + json(res, 200, { ok: true }); + return; + } + + if (req.method === "GET" && pathname === "/ops") { + json(res, 200, { ops: FLOW_OP_SPECS }); + return; + } + + const match = pathname.match(/^\/ops\/([a-z_]+)$/); + if (!match || req.method !== "POST") { + json(res, 404, { error: "Not found" }); + return; + } + + const opName = match[1]; + if (!opNames.has(opName)) { + json(res, 404, { error: `Unknown op: "${opName}"` }); + return; + } + + let params: unknown = {}; + try { + const rawBody = await readBody(req); + if (rawBody) params = JSON.parse(rawBody); + } catch { + json(res, 400, { error: "Invalid JSON body" }); + return; + } + + try { + const result = await ops.execute(opName, params); + json(res, 200, result); + } catch (err) { + log.error( + `[clawflow] manage op "${opName}" error: ${err instanceof Error ? err.message : String(err)}`, + ); + json(res, 500, { error: err instanceof Error ? err.message : String(err) }); + } + }); + + activeServer = server; + server.on("close", () => { + if (activeServer === server) activeServer = null; + }); + + server.on("error", (err: NodeJS.ErrnoException) => { + if (err.code === "EADDRINUSE") { + log.warn( + `[clawflow] port ${port} already in use — skipping management server (another clawflow instance likely owns it)`, + ); + activeServer = null; + return; + } + log.error(`[clawflow] management server error: ${err.message}`); + }); + + // Loopback only — this surface must never be reachable off-box. + server.listen(port, "127.0.0.1", () => { + const addr = server.address(); + const boundPort = typeof addr === "object" && addr ? addr.port : port; + log.info(`[clawflow] management API listening on 127.0.0.1:${boundPort}/ops`); + }); + + return server; +} diff --git a/src/core/op-specs.ts b/src/core/op-specs.ts new file mode 100644 index 0000000..35a6335 --- /dev/null +++ b/src/core/op-specs.ts @@ -0,0 +1,440 @@ +import type { FlowOpSpec } from "./ops.js"; + +// ---- Flow op specs -------------------------------------------------------------- +// The tool surface as data: one spec per flow_* operation, carrying the name, +// description, and JSON-schema parameters. Harness wrappers iterate this to +// register tools (the OpenClaw plugin in-process; out-of-process shims via the +// management API's GET /ops), so the surface is defined exactly once. + +export const FLOW_OP_SPECS: FlowOpSpec[] = [ + { + name: "flow_create", + description: `Create a new clawflow definition and save it to a JSON file. + +Builds a FlowDefinition from the provided parameters, validates it, and writes +it to disk. Use this to scaffold a new flow file; use flow_edit to modify it +afterwards and flow_run to execute it. + +Node types: + ai, agent, branch, condition, loop, parallel, http, memory, wait, sleep, code, exec + +All nodes require "name" and "do". Templates: {{ outputKey.field }}.`, + parameters: { + type: "object", + required: ["file", "flow", "nodes"], + properties: { + file: { + type: "string", + description: + "Filename or path for the new flow file. Plain names like \"my-flow\" are saved to workspace/flows/my-flow.json. Paths with slashes are resolved relative to the workspace.", + }, + flow: { + type: "string", + description: "Unique flow name", + }, + description: { + type: "string", + description: "Human-readable description of what the flow does", + }, + nodes: { + type: "array", + items: { type: "object", additionalProperties: true }, + description: "Array of node definitions", + }, + inputs: { + type: "object", + additionalProperties: true, + description: + 'Declared inputs the flow expects. Map of name → { type?, required?, description?, default? }. Optional: when omitted, the flow accepts any payload. When present, required inputs are enforced before any node runs.', + }, + env: { + type: "object", + additionalProperties: true, + description: "Environment variable defaults", + }, + version: { + type: "string", + description: 'Semver version string, e.g. "1.0.0"', + }, + }, + }, + }, + + { + name: "flow_delete", + description: `Delete a flow file by moving it to the bin. + +The flow is not permanently removed — it is timestamped and moved to +workspace/.clawflow/bin/ so it can be restored later with flow_restore_from_bin. +Safe for agents to call without fear of data loss.`, + parameters: { + type: "object", + required: ["file"], + properties: { + file: { + type: "string", + description: + "Filename or path of the flow to delete. Plain names like \"my-flow\" resolve to workspace/flows/my-flow.json.", + }, + }, + }, + }, + + { + name: "flow_restore_from_bin", + description: `Restore a flow from the bin or list bin contents. + +Without "name", lists all flows in workspace/.clawflow/bin/ with their timestamps. +With "name", restores the most recent version of that flow back to the flows/ +directory. If the flow file already exists, the restore is rejected.`, + parameters: { + type: "object", + properties: { + name: { + type: "string", + description: + "Flow name to restore (without timestamp). Omit to list all bin contents.", + }, + }, + }, + }, + + { + name: "flow_run", + catalogMode: "direct-only", + description: `Run an agentic workflow in the clawflow format. + +State model: + The "input" parameter becomes "inputs" in flow state (i.e. state.inputs). + Flow state = { inputs, env?, ...nodeOutputs }. + Each node with "output" adds its result to state (e.g. output: "result" → state.result). + In code nodes: fn(input, state) — "input" is the resolved node.input field, "state" is the full flow state. + IMPORTANT: inputs contains ALL initial data. If you need different parts of inputs in a code node, + use object-style input: { "payload": "inputs.payload", "email": "inputs.email_to" } + or access via state.inputs.field inside the code. + +Node types: + ai — LLM call, structured or freeform. Use schema: for typed output. + agent — delegate to a real OpenClaw agent. Fields mirror the CLI flags: agent + (a configured slug, e.g. "clawflow" → --agent), sessionKey (a session + key, e.g. "agent:main:slack:channel:agent" → --session-key), sessionId + (→ --session-id), channel (→ --channel). A bare session key is scoped by + agent; an "agent:"-prefixed key is self-scoping. (agentId/session are + deprecated aliases for agent/sessionKey.) + branch — route to different nodes based on a value: { on, paths, default } + loop — iterate over a list: { over, as, nodes[] } + parallel — run nodes concurrently: { nodes[], mode: "all"|"race" } + http — call an external API: { url, method, body, headers } + memory — persist data: { action: read|write|delete, key, value } + wait — pause for approval or event: { for: "approval"|"event", event?, timeout? } + sleep — pause for duration: { duration: "5m" } + code — inline JS expression: { run: "...", input? } + +All nodes support retry: { limit, delay, backoff } and timeout. +Templates: use {{ nodeName.field }} to reference any value in flow state. +Returns instanceId for status tracking and resume. + +Versioning: when a flow has published versions, flow_run uses the latest +published version by default. Set draft: true to run the working copy instead. +Set version to run a specific published version.`, + parameters: { + type: "object", + properties: { + flow: { + type: "object", + properties: { + flow: { type: "string" }, + description: { type: "string" }, + nodes: { + type: "array", + items: { type: "object", additionalProperties: true }, + }, + }, + required: ["flow", "nodes"], + additionalProperties: true, + }, + file: { + type: "string", + description: "Path to a .json flow file (or plain name like \"my-flow\")", + }, + input: { + type: "object", + additionalProperties: true, + description: "Input data, available as inputs.* in the flow (and at state.inputs in code nodes).", + }, + draft: { + type: "boolean", + description: + "Run the draft (working copy) instead of the latest published version. Default: false.", + }, + version: { + type: "number", + description: + "Run a specific published version number. Overrides draft flag.", + }, + }, + }, + }, + + { + name: "flow_resume", + description: `Resume a paused clawflow after an approval gate. +Use the instanceId (= resumeToken) from a flow_run result where status was "paused". +Set approved=true to continue, false to cancel. +You must pass the original flow definition back so the runner can continue.`, + parameters: { + type: "object", + required: ["instanceId", "approved", "flow"], + properties: { + instanceId: { + type: "string", + description: "The instanceId from the paused flow_run result", + }, + approved: { type: "boolean" }, + flow: { + type: "object", + required: ["flow", "nodes"], + properties: { + flow: { type: "string" }, + nodes: { + type: "array", + items: { type: "object", additionalProperties: true }, + }, + }, + additionalProperties: true, + }, + }, + }, + }, + + { + name: "flow_send_event", + description: `Send an event to a flow instance that is waiting with do: wait / for: event. +This is the equivalent of Cloudflare's instance.sendEvent(). +The eventType must match the "event" field on the wait node. +payload is passed as the output of the wait node and into flow state.`, + parameters: { + type: "object", + required: ["instanceId", "eventType"], + properties: { + instanceId: { type: "string" }, + eventType: { + type: "string", + description: "Must match the 'event' field of the wait node", + }, + payload: { + type: "object", + additionalProperties: true, + }, + }, + }, + }, + + { + name: "flow_status", + description: `Get the status and state of a flow instance, or list all instances. +Status values: running | completed | paused | waiting | failed | cancelled`, + parameters: { + type: "object", + properties: { + instanceId: { + type: "string", + description: "Specific instance to inspect. Omit to list all.", + }, + filter: { + type: "string", + description: + "Filter by status: running | completed | paused | waiting | failed | cancelled", + }, + }, + }, + }, + + { + name: "flow_list", + catalogMode: "direct-only", + description: `List all saved flow definitions in the workspace. + +Scans the flows directory for .json files and returns a summary of each flow +including its name, description, declared inputs, version, node count, and file path. +Use this to discover available flows before running or editing them.`, + parameters: { + type: "object", + properties: { + dir: { + type: "string", + description: + "Directory to scan. Defaults to workspace/flows/. Absolute paths are used as-is; relative paths resolve from the workspace root.", + }, + }, + }, + }, + + { + name: "flow_read", + description: `Read a flow definition from file and return its contents. + +Returns the full flow definition (or a single node if specified). The response +includes the declared "inputs" block when present, plus a best-effort list of +input fields referenced by templates (extracted from {{ inputs.* }} usages). +Use this to inspect a flow before running it or to understand what inputs it +needs. + +Versioning: by default reads the draft (working copy). Set version to read +a specific published version. The response includes available version numbers.`, + parameters: { + type: "object", + required: ["file"], + properties: { + file: { + type: "string", + description: + "Filename or path to the flow file. Plain names resolve to workspace/flows/.json.", + }, + node: { + type: "string", + description: + "Name of a specific node to return. Searches nested structures (branches, loops, etc.).", + }, + version: { + type: "number", + description: + "Read a specific published version instead of the draft.", + }, + }, + }, + }, + + { + name: "flow_publish", + description: `Publish the current draft of a flow as a new numbered version. + +Validates the draft, assigns the next version number (auto-incrementing integer), +and saves an immutable copy to .clawflow/versions//.json. +After publishing, flow_run will use this version by default. + +Use this when a flow is ready for production. Edits via flow_edit continue to +modify the draft without affecting published versions.`, + parameters: { + type: "object", + required: ["file"], + properties: { + file: { + type: "string", + description: + "Filename or path to the draft flow file. Plain names resolve to workspace/flows/.json.", + }, + }, + }, + }, + + { + name: "flow_edit", + description: `Edit nodes in a clawflow definition. Operates on a file or inline flow. + +Actions: + set — set top-level flow fields (description, inputs, env, version) + update — update a node entirely or patch specific fields by node name + add — insert a new node at a position (default: end) + remove — remove a node by name + move — move a node to a new position (same level or into a parent) + wrap — wrap one or more nodes into a new container (loop, condition, branch, parallel) + revert — undo the last edit (restores the previous version from history) + list — list all nodes with index, name, type, and output key + +All actions that target nodes (update, remove, move, add) search recursively +through nested structures (branch paths, condition then/else, loop, parallel). + +The "parent" parameter targets a nested node list using slash-separated paths: + "myBranch/true" → branch "myBranch", path "true" + "myCond/then" → condition "myCond", then block + "myLoop" → loop "myLoop", child nodes + "outer/true/inner" → chained nesting + +The edited flow is validated after every mutation. If validation fails, the +edit is rejected and errors are returned. For file-based flows, the file is +overwritten with the updated definition on success. + +Examples: + Set flow fields: { action: "set", fields: { description: "New desc", inputs: { ticket_id: { type: "string", required: true } } } } + Update one field: { action: "update", node: "classify", fields: { prompt: "New prompt" } } + Replace full node: { action: "update", node: "classify", replace: { name: "classify", do: "ai", prompt: "..." } } + Add at position: { action: "add", position: 2, nodeDefinition: { name: "step3", do: "code", run: "..." } } + Add inside branch: { action: "add", parent: "shouldUpdate/true", nodeDefinition: { name: "step3", do: "code", run: "..." } } + Remove: { action: "remove", node: "old-step" } + Move into loop: { action: "move", node: "step3", parent: "myLoop", position: 0 } + Move into else: { action: "move", node: "step3", parent: "myCond/else", position: 0 } + Wrap in loop: { action: "wrap", nodes: ["step1", "step2"], wrapper: { name: "myLoop", do: "loop", over: "items", as: "item" } } + Wrap in condition: { action: "wrap", nodes: ["step1", "step2"], wrapper: { name: "guard", do: "condition", if: "{{ items.length > 0 }}" } } + List: { action: "list" } + +To RESTRUCTURE an existing flow — make a group of nodes conditional, nest them in a +loop, or reparent them — use "wrap" (many nodes at once) or "move" (one node into a +"parent" path like "guard/else"). Do NOT remove-and-re-add each node, and do NOT +re-send the whole flow: a single large edit is slow and can stall mid-generation.`, + parameters: { + type: "object", + required: ["action"], + properties: { + file: { + type: "string", + description: "Path to a .json flow file. Mutually exclusive with 'flow'.", + }, + flow: { + type: "object", + description: "Inline flow definition. Mutually exclusive with 'file'.", + properties: { + flow: { type: "string" }, + nodes: { + type: "array", + items: { type: "object", additionalProperties: true }, + }, + }, + required: ["flow", "nodes"], + additionalProperties: true, + }, + action: { + type: "string", + enum: ["set", "update", "add", "remove", "move", "wrap", "revert", "list"], + }, + node: { + type: "string", + description: "Node name to target (required for update, remove, move). Searched recursively through nested structures.", + }, + fields: { + type: "object", + additionalProperties: true, + description: "For action=set: top-level flow fields to set (description, inputs, env, version). For action=update: partial field updates to merge into the node.", + }, + replace: { + type: "object", + additionalProperties: true, + description: "For action=update: full node replacement (must include name and do)", + }, + nodeDefinition: { + type: "object", + additionalProperties: true, + description: "For action=add: the new node definition", + }, + position: { + type: "number", + description: "For action=add/move: index to insert/move to (0-based). Default: end.", + }, + parent: { + type: "string", + description: 'For action=add/move: target a nested node list. Slash-separated path e.g. "myBranch/true", "myCond/then", "myLoop". For move: destination parent (node is removed from current location and inserted here).', + }, + nodes: { + type: "array", + items: { type: "string" }, + description: "For action=wrap: array of node names to wrap into a container.", + }, + wrapper: { + type: "object", + additionalProperties: true, + description: "For action=wrap: the container node definition (must include name, do). Wrapped nodes become its children (e.g. loop.nodes, condition.then, branch.paths.true, parallel.nodes).", + }, + }, + }, + }, +]; diff --git a/src/core/ops.ts b/src/core/ops.ts new file mode 100644 index 0000000..6d07a23 --- /dev/null +++ b/src/core/ops.ts @@ -0,0 +1,1091 @@ +import * as fs from "node:fs"; +import * as path from "node:path"; + +import { FlowRunner, sendEvent } from "./runner.js"; +import { validateFlow } from "./validate.js"; +import type { + FlowDefinition, + FlowNode, + BranchNode, + ConditionNode, + LoopNode, + ParallelNode, +} from "./types.js"; + +// ---- Flow operations ------------------------------------------------------------ +// The harness-neutral implementation of the flow_* tool surface. Each operation +// returns a plain { text, details? } result; harness wrappers (the OpenClaw +// plugin, the management API, future harness shims) adapt that to their own +// tool-result format. FLOW_OP_SPECS carries the tool metadata (name, +// description, JSON-schema parameters) as data so every harness registers the +// same surface without duplicating it. + +export interface OpResult { + text: string; + details?: unknown; +} + +export interface FlowOpSpec { + /** Tool name, e.g. "flow_create". Also the op name for FlowOps.execute(). */ + name: string; + description: string; + /** JSON-schema for the tool parameters. */ + parameters: object; + /** OpenClaw catalog hint — passed through by the OpenClaw plugin only. */ + catalogMode?: string; +} + +// ---- Shared helpers ------------------------------------------------------------- + +/** + * Recursively extract unique inputs.* paths from any string values in an + * object tree. This tells agents what input fields a flow references — used + * when the flow has no declared inputs block to give a best-effort hint. + */ +function extractInputRefs(obj: unknown): string[] { + const paths = new Set(); + const re = /\{\{\s*inputs\.(\w+(?:\.\w+)*)/g; + + function walk(val: unknown): void { + if (typeof val === "string") { + let m: RegExpExecArray | null; + while ((m = re.exec(val)) !== null) paths.add(m[1]); + } else if (Array.isArray(val)) { + for (const item of val) walk(item); + } else if (val && typeof val === "object") { + for (const v of Object.values(val)) walk(v); + } + } + walk(obj); + return [...paths].sort(); +} + +/** Find a node by name, searching nested structures (branch paths, loop, parallel, condition). */ +function findNode(nodes: FlowNode[], name: string): FlowNode | undefined { + for (const n of nodes) { + if (n.name === name) return n; + if ("paths" in n && n.paths) { + for (const branch of Object.values( + n.paths as Record, + )) { + const found = findNode(branch, name); + if (found) return found; + } + } + if ("default" in n && Array.isArray(n.default)) { + const found = findNode(n.default as FlowNode[], name); + if (found) return found; + } + if ("nodes" in n && Array.isArray(n.nodes)) { + const found = findNode(n.nodes as FlowNode[], name); + if (found) return found; + } + if ("then" in n && Array.isArray(n.then)) { + const found = findNode(n.then as FlowNode[], name); + if (found) return found; + } + if ("else" in n && Array.isArray((n as ConditionNode).else)) { + const found = findNode((n as ConditionNode).else!, name); + if (found) return found; + } + } + return undefined; +} + +// ---- FlowOps -------------------------------------------------------------------- + +export class FlowOps { + constructor( + private workspace: string, + private runner: FlowRunner, + ) {} + + private get store() { + return this.runner.getStore(); + } + + /** Resolve a file param to an absolute path using workspace conventions. */ + private resolveFlowFile(file: string): string { + const base = this.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. */ + private versionsDir(flowName: string): string { + return path.join(this.workspace, ".clawflow", "versions", flowName); + } + + /** List all published version numbers for a flow, sorted ascending. */ + private listVersions(flowName: string): number[] { + const dir = this.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); + } + + /** Read a specific published version. Returns null if not found. */ + private readVersion(flowName: string, version: number): FlowDefinition | null { + const file = path.join(this.versionsDir(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. */ + private readLatestVersion(flowName: string): { version: number; def: FlowDefinition } | null { + const versions = this.listVersions(flowName); + if (versions.length === 0) return null; + const latest = versions[versions.length - 1]; + const def = this.readVersion(flowName, latest); + if (!def) return null; + return { version: latest, def }; + } + + /** Dispatch an op by its tool name (e.g. "flow_run"). Throws on unknown name. */ + async execute(name: string, params: unknown): Promise { + const p = (params ?? {}) as never; + switch (name) { + case "flow_create": return this.create(p); + case "flow_delete": return this.deleteFlow(p); + case "flow_restore_from_bin": return this.restoreFromBin(p); + case "flow_run": return this.run(p); + case "flow_resume": return this.resume(p); + case "flow_send_event": return this.sendEvent(p); + case "flow_status": return this.status(p); + case "flow_list": return this.list(p); + case "flow_read": return this.read(p); + case "flow_publish": return this.publish(p); + case "flow_edit": return this.edit(p); + default: + throw new Error(`Unknown flow op: "${name}"`); + } + } + + // ---- flow_create -------------------------------------------------------------- + + async create(params: { + file: string; + flow: string; + description?: string; + nodes: FlowNode[]; + inputs?: FlowDefinition["inputs"]; + env?: Record; + version?: string; + }): Promise { + const abs = this.resolveFlowFile(params.file); + + if (fs.existsSync(abs)) + return { text: `File already exists: ${abs}. Use flow_edit to modify it.` }; + + const flowDef: FlowDefinition = { + flow: params.flow, + ...(params.version && { version: params.version }), + ...(params.description && { description: params.description }), + ...(params.inputs && { inputs: params.inputs }), + ...(params.env && { env: params.env }), + nodes: params.nodes, + }; + + const validation = validateFlow(flowDef); + if (!validation.ok) + return { + text: `Validation failed:\n${validation.errors.map((e) => ` - ${e.message}`).join("\n")}`, + }; + + try { + fs.mkdirSync(path.dirname(abs), { recursive: true }); + } catch (err) { + const msg = err instanceof Error ? err.message : String(err); + return { + text: `Cannot create directory for ${abs}: ${msg}. Is OPENCLAW_WORKSPACE set? Current workspace root: ${this.workspace}`, + }; + } + fs.writeFileSync(abs, JSON.stringify(flowDef, null, 2) + "\n"); + + return { + text: `Flow "${params.flow}" created at ${abs}`, + details: flowDef, + }; + } + + // ---- flow_delete -------------------------------------------------------------- + + async deleteFlow(params: { file: string }): Promise { + const abs = this.resolveFlowFile(params.file); + + if (!fs.existsSync(abs)) + return { text: `File not found: ${abs}` }; + + const binDir = path.join(this.workspace, ".clawflow", "bin"); + fs.mkdirSync(binDir, { recursive: true }); + + const basename = path.basename(abs, ".json"); + const ts = new Date().toISOString().replace(/[:.]/g, "-"); + const binPath = path.join(binDir, `${basename}.${ts}.json`); + + fs.renameSync(abs, binPath); + + return { text: `Flow moved to bin: ${binPath}` }; + } + + // ---- flow_restore_from_bin ------------------------------------------------------ + + async restoreFromBin(params: { name?: string }): Promise { + const binDir = path.join(this.workspace, ".clawflow", "bin"); + + if (!fs.existsSync(binDir)) + return { text: "Bin is empty." }; + + const files = fs.readdirSync(binDir) + .filter((f: string) => f.endsWith(".json")) + .sort() + .reverse(); + + if (files.length === 0) + return { text: "Bin is empty." }; + + // ---- List mode --- + if (!params.name) { + const entries = files.map((f: string) => { + const match = f.match(/^(.+?)\.(\d{4}-.+)\.json$/); + return { + name: match?.[1] ?? f, + deletedAt: match?.[2] ?? "unknown", + file: f, + }; + }); + return { + text: JSON.stringify(entries, null, 2), + details: entries, + }; + } + + // ---- Restore mode --- + const prefix = params.name.replace(/\.json$/, ""); + const match = files.find((f: string) => f.startsWith(`${prefix}.`)); + + if (!match) + return { text: `No bin entry found for "${prefix}".` }; + + const dest = path.join(this.workspace, "flows", `${prefix}.json`); + if (fs.existsSync(dest)) + return { + text: `Cannot restore: ${dest} already exists. Delete or rename it first.`, + }; + + fs.renameSync(path.join(binDir, match), dest); + + return { text: `Restored "${prefix}" from bin to ${dest}` }; + } + + // ---- flow_run ----------------------------------------------------------------- + + async run(params: { + flow?: FlowDefinition; + file?: string; + input?: unknown; + draft?: boolean; + version?: number; + }): Promise { + let flowDef: FlowDefinition; + let source = ""; + + if (params.file) { + const abs = this.resolveFlowFile(params.file); + + // Determine flow name for version lookup + const flowName = path.basename(abs, ".json"); + + if (params.version != null) { + // Specific version requested + const vDef = this.readVersion(flowName, params.version); + if (!vDef) + return { + text: `Version ${params.version} not found for flow "${flowName}". Available: ${this.listVersions(flowName).join(", ") || "none (not yet published)"}`, + }; + flowDef = vDef; + source = `v${params.version}`; + } else if (!params.draft) { + // Default: use latest published version if available + const latest = this.readLatestVersion(flowName); + if (latest) { + flowDef = latest.def; + source = `v${latest.version}`; + } else { + // No published versions — fall back to draft + if (!fs.existsSync(abs)) + return { text: `File not found: ${abs}` }; + flowDef = JSON.parse(fs.readFileSync(abs, "utf8")) as FlowDefinition; + source = "draft (no published versions)"; + } + } else { + // Explicit draft mode + if (!fs.existsSync(abs)) + return { text: `File not found: ${abs}` }; + flowDef = JSON.parse(fs.readFileSync(abs, "utf8")) as FlowDefinition; + source = "draft"; + } + } else if (params.flow) { + flowDef = params.flow as FlowDefinition; + source = "inline"; + } else { + return { text: "Provide either `flow` (inline definition) or `file` (path)." }; + } + + try { + const result = await this.runner.run(flowDef, params.input ?? {}); + const out = { ...result, _source: source }; + return { + text: JSON.stringify(out, null, 2), + details: out, + }; + } catch (err) { + return { + text: `Error: ${err instanceof Error ? err.message : String(err)}`, + }; + } + } + + // ---- flow_resume -------------------------------------------------------------- + + async resume(params: { + instanceId: string; + approved: boolean; + flow: FlowDefinition; + }): Promise { + try { + const result = await this.runner.resume( + params.instanceId, + params.flow, + params.approved, + ); + return { + text: JSON.stringify(result, null, 2), + details: result, + }; + } catch (err) { + return { + text: `Error: ${err instanceof Error ? err.message : String(err)}`, + }; + } + } + + // ---- flow_send_event ---------------------------------------------------------- + + async sendEvent(params: { + instanceId: string; + eventType: string; + payload?: unknown; + }): Promise { + const delivered = sendEvent( + params.instanceId, + params.eventType, + params.payload ?? {}, + ); + if (!delivered) { + return { + text: `No active waiter found for instance "${params.instanceId}" event "${params.eventType}". The flow may not be waiting, or the event type doesn't match.`, + }; + } + return { + text: `Event "${params.eventType}" delivered to instance "${params.instanceId}".`, + }; + } + + // ---- flow_status -------------------------------------------------------------- + + async status(params: { instanceId?: string; filter?: string }): Promise { + if (params.instanceId) { + const record = this.store.get(params.instanceId); + if (!record) + return { text: `Instance not found: ${params.instanceId}` }; + return { + text: JSON.stringify(record, null, 2), + details: record, + }; + } + const records = this.store.list(params.filter); + const summary = records.map((r) => ({ + instanceId: r.instanceId, + flow: r.flowName, + status: r.status, + updatedAt: r.updatedAt, + waitingFor: r.waitingFor, + })); + return { + text: JSON.stringify(summary, null, 2), + details: summary, + }; + } + + // ---- flow_list ---------------------------------------------------------------- + + async list(params: { dir?: string }): Promise { + const base = this.workspace; + + let dir: string; + if (!params.dir) { + dir = path.join(base, "flows"); + } else if (params.dir.startsWith("/")) { + dir = params.dir; + } else { + dir = path.join(base, params.dir); + } + + if (!fs.existsSync(dir)) { + return { text: `Flows directory not found: ${dir}` }; + } + + const files = fs.readdirSync(dir).filter((f: string) => f.endsWith(".json")); + + if (files.length === 0) { + return { text: `No flow files found in ${dir}` }; + } + + const flows: Array<{ + file: string; + flow: string; + description?: string; + inputs?: FlowDefinition["inputs"]; + expectedInputs?: string[]; + publishedVersion?: number; + totalVersions?: number; + nodes: number; + }> = []; + + for (const file of files) { + const abs = path.join(dir, file); + try { + const raw = fs.readFileSync(abs, "utf-8"); + const def = JSON.parse(raw) as FlowDefinition; + if (!def.flow || !Array.isArray(def.nodes)) continue; + const referenced = extractInputRefs(def.nodes); + const flowName = file.replace(/\.json$/, ""); + const versions = this.listVersions(flowName); + flows.push({ + file: abs, + flow: def.flow, + ...(def.description && { description: def.description }), + ...(def.inputs && { inputs: def.inputs }), + ...(referenced.length > 0 && { expectedInputs: referenced }), + ...(versions.length > 0 && { + publishedVersion: versions[versions.length - 1], + totalVersions: versions.length, + }), + nodes: def.nodes.length, + }); + } catch { + // skip non-flow JSON files + } + } + + if (flows.length === 0) { + return { text: `No valid flow definitions found in ${dir}` }; + } + + return { + text: JSON.stringify(flows, null, 2), + details: flows, + }; + } + + // ---- flow_read ---------------------------------------------------------------- + + async read(params: { file: string; node?: string; version?: number }): Promise { + const abs = this.resolveFlowFile(params.file); + const flowName = path.basename(abs, ".json"); + const versions = this.listVersions(flowName); + + let flowDef: FlowDefinition; + let source: string; + + if (params.version != null) { + const vDef = this.readVersion(flowName, params.version); + if (!vDef) { + return { + text: `Version ${params.version} not found for "${flowName}". Available: ${versions.join(", ") || "none"}`, + }; + } + flowDef = vDef; + source = `v${params.version}`; + } else { + if (!fs.existsSync(abs)) { + return { text: `File not found: ${abs}` }; + } + try { + flowDef = JSON.parse(fs.readFileSync(abs, "utf-8")) as FlowDefinition; + } catch (err) { + return { + text: `Failed to parse ${abs}: ${err instanceof Error ? err.message : String(err)}`, + }; + } + source = "draft"; + } + + const referenced = extractInputRefs(flowDef.nodes); + + // Single node mode + if (params.node) { + const found = findNode(flowDef.nodes, params.node); + if (!found) { + return { + text: `Node "${params.node}" not found in flow "${flowDef.flow}". Nodes: ${flowDef.nodes.map((n) => n.name).join(", ")}`, + }; + } + const nodeInputs = extractInputRefs(found); + const result = { + flow: flowDef.flow, + _source: source, + _file: abs, + ...(versions.length > 0 && { _versions: versions }), + ...(nodeInputs.length > 0 && { expectedInputs: nodeInputs }), + node: found, + }; + return { + text: JSON.stringify(result, null, 2), + details: result, + }; + } + + // Full flow mode + const result = { + ...flowDef, + _source: source, + _file: abs, + ...(versions.length > 0 && { _versions: versions }), + ...(referenced.length > 0 && { _expectedInputs: referenced }), + }; + return { + text: JSON.stringify(result, null, 2), + details: result, + }; + } + + // ---- flow_publish -------------------------------------------------------------- + + async publish(params: { file: string }): Promise { + const abs = this.resolveFlowFile(params.file); + if (!fs.existsSync(abs)) { + return { text: `Draft not found: ${abs}` }; + } + + let flowDef: FlowDefinition; + try { + flowDef = JSON.parse(fs.readFileSync(abs, "utf-8")) as FlowDefinition; + } catch (err) { + return { + text: `Failed to parse ${abs}: ${err instanceof Error ? err.message : String(err)}`, + }; + } + + const validation = validateFlow(flowDef); + if (!validation.ok) { + return { + text: `Validation failed — cannot publish:\n${validation.errors.map((e) => ` - ${e.message}`).join("\n")}`, + }; + } + + const flowName = path.basename(abs, ".json"); + const versions = this.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 = this.versionsDir(flowName); + fs.mkdirSync(dir, { recursive: true }); + const versionFile = path.join(dir, `${nextVersion}.json`); + fs.writeFileSync(versionFile, JSON.stringify(flowDef, null, 2) + "\n"); + + return { + text: `Published "${flowDef.flow}" as v${nextVersion}. flow_run will now use this version by default.\nFile: ${versionFile}`, + details: { + flow: flowDef.flow, + version: nextVersion, + file: versionFile, + totalVersions: nextVersion, + }, + }; + } + + // ---- flow_edit ---------------------------------------------------------------- + + async edit(params: { + file?: string; + flow?: FlowDefinition; + action: "set" | "update" | "add" | "remove" | "move" | "wrap" | "revert" | "list"; + node?: string; + fields?: Record; + replace?: FlowNode; + nodeDefinition?: FlowNode; + position?: number; + parent?: string; + nodes?: string[]; + wrapper?: FlowNode; + }): Promise { + // ---- Load flow definition --- + let flowDef: FlowDefinition; + let filePath: string | undefined; + + if (params.file) { + const abs = this.resolveFlowFile(params.file); + if (!fs.existsSync(abs)) + return { text: `File not found: ${abs}` }; + filePath = abs; + flowDef = JSON.parse(fs.readFileSync(abs, "utf8")) as FlowDefinition; + } else if (params.flow) { + flowDef = params.flow; + } else { + return { text: "Provide either `file` (path) or `flow` (inline definition)." }; + } + + const ok = (msg: string, flow: FlowDefinition): OpResult => ({ + text: msg, + details: flow, + }); + const fail = (msg: string): OpResult => ({ text: msg }); + + // Resolve a node list from a parent path. Supports: + // undefined → flowDef.nodes (top-level) + // "myBranch/true" → branch node "myBranch", paths["true"] + // "myBranch/default" → branch node "myBranch", default + // "myCond/then" → condition node "myCond", then + // "myCond/else" → condition node "myCond", else + // "myLoop" → loop node "myLoop", nodes + // "myParallel" → parallel node "myParallel", nodes + // Paths can be chained: "outerBranch/true/innerLoop" + const resolveNodeList = ( + parentPath?: string, + ): { nodes: FlowNode[]; error?: string } => { + if (!parentPath) return { nodes: flowDef.nodes }; + + const parts = parentPath.split("/"); + let current: FlowNode[] = flowDef.nodes; + + for (let i = 0; i < parts.length; i++) { + const part = parts[i]; + const node = current.find((n) => n.name === part); + if (!node) + return { + nodes: [], + error: `Parent node "${part}" not found at level ${i}.`, + }; + + if (node.do === "branch") { + const b = node as BranchNode; + const key = parts[++i]; // consume next part as path key + if (!key) + return { + nodes: [], + error: `Branch "${part}" requires a path key (e.g., "${part}/true").`, + }; + if (key === "default") { + if (!b.default) b.default = []; + current = b.default; + } else { + if (!b.paths[key]) b.paths[key] = []; + current = b.paths[key]; + } + } else if (node.do === "condition") { + const c = node as ConditionNode; + const key = parts[++i]; // consume next part as then/else + if (!key) + return { + nodes: [], + error: `Condition "${part}" requires "then" or "else" (e.g., "${part}/then").`, + }; + if (key === "then") { + current = c.then; + } else if (key === "else") { + if (!c.else) c.else = []; + current = c.else; + } else { + return { + nodes: [], + error: `Condition "${part}" only accepts "then" or "else", got "${key}".`, + }; + } + } else if (node.do === "loop") { + current = (node as LoopNode).nodes; + } else if (node.do === "parallel") { + current = (node as ParallelNode).nodes; + } else { + return { + nodes: [], + error: `Node "${part}" (do: "${node.do}") has no child nodes.`, + }; + } + } + return { nodes: current }; + }; + + // Deep-find: search all node lists recursively, return the containing array + index + const deepFind = ( + name: string, + nodes: FlowNode[] = flowDef.nodes, + ): { list: FlowNode[]; index: number } | null => { + const idx = nodes.findIndex((n) => n.name === name); + if (idx !== -1) return { list: nodes, index: idx }; + for (const node of nodes) { + const children = getChildLists(node); + for (const child of children) { + const found = deepFind(name, child); + if (found) return found; + } + } + return null; + }; + + // Get all child node arrays from a container node + const getChildLists = (node: FlowNode): FlowNode[][] => { + switch (node.do) { + case "branch": { + const b = node as BranchNode; + const lists = Object.values(b.paths); + if (b.default) lists.push(b.default); + return lists; + } + case "condition": { + const c = node as ConditionNode; + const lists: FlowNode[][] = [c.then]; + if (c.else) lists.push(c.else); + return lists; + } + case "loop": + return [(node as LoopNode).nodes]; + case "parallel": + return [(node as ParallelNode).nodes]; + default: + return []; + } + }; + + // Save a snapshot before mutating (file-based flows only). + // Derive history dir from the flow file's parent (sibling to flows/). + const saveSnapshot = () => { + if (!filePath) return; + const flowDir = path.dirname(filePath); + const root = path.dirname(flowDir); // parent of flows/ + const flowName = path.basename(filePath, ".json"); + const histDir = path.join(root, ".clawflow", "history", flowName); + try { + fs.mkdirSync(histDir, { recursive: true }); + const ts = new Date().toISOString().replace(/[:.]/g, "-"); + const snapPath = path.join(histDir, `${ts}.json`); + fs.writeFileSync(snapPath, fs.readFileSync(filePath, "utf8")); + } catch { + // History is best-effort — don't block the edit + } + }; + + // ---- Revert (undo last edit) --- + if (params.action === "revert") { + if (!filePath) + return fail("Revert only works on file-based flows."); + const flowDir = path.dirname(filePath); + const root = path.dirname(flowDir); + const flowName = path.basename(filePath, ".json"); + const histDir = path.join(root, ".clawflow", "history", flowName); + if (!fs.existsSync(histDir)) + return fail("No edit history found for this flow."); + const snaps = fs.readdirSync(histDir) + .filter((f: string) => f.endsWith(".json")) + .sort() + .reverse(); + if (snaps.length === 0) + return fail("No edit history found for this flow."); + const latest = path.join(histDir, snaps[0]); + const restored = JSON.parse(fs.readFileSync(latest, "utf8")) as FlowDefinition; + fs.writeFileSync(filePath, JSON.stringify(restored, null, 2) + "\n"); + fs.unlinkSync(latest); + return ok( + `Reverted to snapshot ${snaps[0]}. File written: ${filePath}`, + restored, + ); + } + + // ---- Set (top-level flow fields) --- + if (params.action === "set") { + if (!params.fields) + return fail('action "set" requires "fields".'); + + const allowed = ["description", "inputs", "env", "version"]; + const backup = { ...flowDef }; + + for (const [key, value] of Object.entries(params.fields)) { + if (!allowed.includes(key)) + return fail( + `Cannot set "${key}". Allowed fields: ${allowed.join(", ")}. Use "update" to modify nodes.`, + ); + if (value === null || value === undefined) { + delete (flowDef as unknown as Record)[key]; + } else { + (flowDef as unknown as Record)[key] = value; + } + } + + const validation = validateFlow(flowDef); + if (!validation.ok) { + // rollback + Object.assign(flowDef, backup); + return fail( + `Validation failed after set:\n${validation.errors.map((e) => ` - ${e.message}`).join("\n")}`, + ); + } + + if (filePath) { + saveSnapshot(); + fs.writeFileSync(filePath, JSON.stringify(flowDef, null, 2) + "\n"); + } + return ok( + `Flow fields updated: ${Object.keys(params.fields).join(", ")}.${filePath ? ` File written: ${filePath}` : ""}`, + flowDef, + ); + } + + // ---- List --- + if (params.action === "list") { + const list = flowDef.nodes.map((n, i) => ({ + index: i, + name: n.name, + do: n.do, + output: n.output ?? null, + })); + return { + text: JSON.stringify(list, null, 2), + details: list, + }; + } + + // ---- Update --- + if (params.action === "update") { + if (!params.node) + return fail('action "update" requires "node" (node name).'); + const found = deepFind(params.node); + if (!found) + return fail(`Node "${params.node}" not found.`); + + if (params.replace) { + found.list[found.index] = params.replace as FlowNode; + } else if (params.fields) { + found.list[found.index] = { + ...found.list[found.index], + ...params.fields, + } as FlowNode; + } else { + return fail( + 'action "update" requires either "fields" (partial) or "replace" (full node).', + ); + } + + const validation = validateFlow(flowDef); + if (!validation.ok) + return fail( + `Validation failed after update:\n${validation.errors.map((e) => ` - ${e.message}`).join("\n")}`, + ); + + if (filePath) { + saveSnapshot(); + fs.writeFileSync(filePath, JSON.stringify(flowDef, null, 2) + "\n"); + } + return ok( + `Node "${params.node}" updated.${filePath ? ` File written: ${filePath}` : ""}`, + flowDef, + ); + } + + // ---- Add --- + if (params.action === "add") { + if (!params.nodeDefinition) + return fail('action "add" requires "nodeDefinition".'); + const { nodes: targetList, error: parentErr } = resolveNodeList(params.parent); + if (parentErr) return fail(parentErr); + const pos = params.position ?? targetList.length; + if (pos < 0 || pos > targetList.length) + return fail( + `Position ${pos} out of range (0-${targetList.length}).`, + ); + targetList.splice(pos, 0, params.nodeDefinition as FlowNode); + + const validation = validateFlow(flowDef); + if (!validation.ok) { + targetList.splice(pos, 1); // rollback + return fail( + `Validation failed after add:\n${validation.errors.map((e) => ` - ${e.message}`).join("\n")}`, + ); + } + + if (filePath) { + saveSnapshot(); + fs.writeFileSync(filePath, JSON.stringify(flowDef, null, 2) + "\n"); + } + const loc = params.parent ? ` in ${params.parent}` : ""; + return ok( + `Node "${params.nodeDefinition.name}" added at position ${pos}${loc}.${filePath ? ` File written: ${filePath}` : ""}`, + flowDef, + ); + } + + // ---- Remove --- + if (params.action === "remove") { + if (!params.node) + return fail('action "remove" requires "node" (node name).'); + const found = deepFind(params.node); + if (!found) + return fail(`Node "${params.node}" not found.`); + + const removed = found.list.splice(found.index, 1)[0]; + + const validation = validateFlow(flowDef); + if (!validation.ok) { + found.list.splice(found.index, 0, removed); // rollback + return fail( + `Validation failed after remove (rolled back):\n${validation.errors.map((e) => ` - ${e.message}`).join("\n")}`, + ); + } + + if (filePath) { + saveSnapshot(); + fs.writeFileSync(filePath, JSON.stringify(flowDef, null, 2) + "\n"); + } + return ok( + `Node "${params.node}" removed.${filePath ? ` File written: ${filePath}` : ""}`, + flowDef, + ); + } + + // ---- Move --- + if (params.action === "move") { + if (!params.node) + return fail('action "move" requires "node" (node name).'); + if (params.position === undefined) + return fail('action "move" requires "position".'); + + // Find the node wherever it is + const source = deepFind(params.node); + if (!source) + return fail(`Node "${params.node}" not found.`); + + // Resolve destination list + const { nodes: destList, error: parentErr } = resolveNodeList(params.parent); + if (parentErr) return fail(parentErr); + + // Remove from source + const [moved] = source.list.splice(source.index, 1); + const pos = Math.min(params.position, destList.length); + destList.splice(pos, 0, moved); + + const validation = validateFlow(flowDef); + if (!validation.ok) { + // rollback: remove from dest, re-insert at source + destList.splice(pos, 1); + source.list.splice(source.index, 0, moved); + return fail( + `Validation failed after move (rolled back):\n${validation.errors.map((e) => ` - ${e.message}`).join("\n")}`, + ); + } + + if (filePath) { + saveSnapshot(); + fs.writeFileSync(filePath, JSON.stringify(flowDef, null, 2) + "\n"); + } + const loc = params.parent ? ` in ${params.parent}` : ""; + return ok( + `Node "${params.node}" moved to position ${pos}${loc}.${filePath ? ` File written: ${filePath}` : ""}`, + flowDef, + ); + } + + // ---- Wrap --- + if (params.action === "wrap") { + if (!params.nodes || params.nodes.length === 0) + return fail('action "wrap" requires "nodes" (array of node names).'); + if (!params.wrapper) + return fail('action "wrap" requires "wrapper" (container node definition with name and do).'); + if (!["loop", "condition", "branch", "parallel"].includes(params.wrapper.do)) + return fail( + `Wrapper "do" must be loop, condition, branch, or parallel. Got "${params.wrapper.do}".`, + ); + + // All target nodes must be in the same parent list and contiguous + const firstFound = deepFind(params.nodes[0]); + if (!firstFound) + return fail(`Node "${params.nodes[0]}" not found.`); + + const parentList = firstFound.list; + const indices: number[] = []; + for (const name of params.nodes) { + const idx = parentList.findIndex((n) => n.name === name); + if (idx === -1) + return fail( + `Node "${name}" not found in the same parent list as "${params.nodes[0]}".`, + ); + indices.push(idx); + } + indices.sort((a, b) => a - b); + + // Check contiguous + for (let i = 1; i < indices.length; i++) { + if (indices[i] !== indices[i - 1] + 1) + return fail( + "Nodes to wrap must be contiguous (adjacent in the same list).", + ); + } + + // Extract the nodes + const extracted = parentList.splice(indices[0], indices.length); + + // Build the wrapper node with children + const wrapperNode = { ...params.wrapper } as FlowNode; + switch (wrapperNode.do) { + case "loop": + (wrapperNode as LoopNode).nodes = extracted; + break; + case "parallel": + (wrapperNode as ParallelNode).nodes = extracted; + break; + case "condition": + (wrapperNode as ConditionNode).then = extracted; + if (!(wrapperNode as ConditionNode).else) + (wrapperNode as ConditionNode).else = []; + break; + case "branch": { + const b = wrapperNode as BranchNode; + // Put extracted nodes into the first path, or "true" by default + const firstPath = b.paths ? Object.keys(b.paths)[0] : "true"; + if (!b.paths) b.paths = {}; + b.paths[firstPath] = extracted; + break; + } + } + + // Insert wrapper where the first extracted node was + parentList.splice(indices[0], 0, wrapperNode); + + const validation = validateFlow(flowDef); + if (!validation.ok) { + // rollback: remove wrapper, re-insert extracted nodes + parentList.splice(indices[0], 1); + parentList.splice(indices[0], 0, ...extracted); + return fail( + `Validation failed after wrap (rolled back):\n${validation.errors.map((e) => ` - ${e.message}`).join("\n")}`, + ); + } + + if (filePath) { + saveSnapshot(); + fs.writeFileSync(filePath, JSON.stringify(flowDef, null, 2) + "\n"); + } + return ok( + `Wrapped ${params.nodes.length} node(s) into "${params.wrapper.name}" (${params.wrapper.do}).${filePath ? ` File written: ${filePath}` : ""}`, + flowDef, + ); + } + + return fail(`Unknown action: "${params.action}"`); + } +} diff --git a/src/core/runner.ts b/src/core/runner.ts index f883f7c..45fb4d6 100644 --- a/src/core/runner.ts +++ b/src/core/runner.ts @@ -31,6 +31,7 @@ import { } from "./types.js"; import { StateStore } from "./store.js"; import { validateFlow } from "./validate.js"; +import { OpenClawCliInvoker, type AgentInvoker } from "./agent-invoker.js"; import { defaultRegistry, type CustomStepContext, @@ -116,11 +117,14 @@ export class FlowRunner { * flow-cancel API is wired in, it should call abort() and delete here. */ private abortControllers = new Map(); + private agentInvoker: AgentInvoker; constructor(cfg: PluginConfig) { this.cfg = cfg; this.store = new StateStore(cfg.stateDir); this.registry = cfg.customSteps ?? defaultRegistry; + this.agentInvoker = + cfg.agentInvoker ?? new OpenClawCliInvoker({ defaultAgent: cfg.defaultAgent }); } // ---- Start a new run ---------------------------------------------------------- @@ -968,8 +972,8 @@ export class FlowRunner { } // ---- do: agent ---------------------------------------------------------------- - // Delegates to a real OpenClaw agent via CLI. The agent gets full tool access - // (browser, exec, memory, MCP, CLI). + // Delegates to a real agent via the configured AgentInvoker (default: the + // OpenClaw CLI). The agent gets full tool access (browser, exec, memory, MCP, CLI). private async execAgent( node: AgentNode, @@ -986,7 +990,7 @@ export class FlowRunner { ? parseDuration(node.timeout) : undefined; - const cliResult = await this.tryOpenClawAgent( + const result = await this.agentInvoker.invoke( fullPrompt, { // new names win; agentId/session are accepted as deprecated aliases @@ -995,77 +999,12 @@ export class FlowRunner { sessionId: node.sessionId, channel: node.channel, }, - state, - timeoutMs, + { + timeoutMs: timeoutMs ?? this.cfg.maxNodeDurationMs ?? 120_000, + env: (state.env as Record) ?? {}, + }, ); - return { output: this.autoParseJson(cliResult) }; - } - - // Build the `openclaw agent` selector args from a node's target fields. Mirrors - // OpenClaw's own selectors, which compose: - // --agent scopes to a configured agent - // --session-key an exact session key (agent::, or scoped to --agent) - // --session-id the most specific: one explicit session (scoped by --agent) - // --channel delivery channel for the reply - // We pass through whichever are set and let OpenClaw resolve/enforce consistency - // rather than re-asserting it here. When no target selector is set at all, fall - // back to a configured agent (defaultAgent > "main") so the call is routable. - private buildAgentArgs(target: { - agent?: string; - sessionKey?: string; - sessionId?: string; - channel?: string; - }): string[] { - const { agent, sessionKey, sessionId, channel } = target; - const args: string[] = []; - if (agent) args.push("--agent", agent); - if (sessionKey) args.push("--session-key", sessionKey); - if (sessionId) args.push("--session-id", sessionId); - if (channel) args.push("--channel", channel); - if (!agent && !sessionKey && !sessionId) { - args.push("--agent", this.cfg.defaultAgent ?? "main"); - } - return args; - } - - private async tryOpenClawAgent( - message: string, - target: { agent?: string; sessionKey?: string; sessionId?: string; channel?: string }, - state: FlowState, - nodeTimeoutMs?: number, - ): Promise { - const { execFile } = await import("child_process"); - const { promisify } = await import("util"); - const execFileAsync = promisify(execFile); - - // Check if openclaw CLI is available - try { - await execFileAsync("which", ["openclaw"]); - } catch { - throw new Error("openclaw CLI not found — agent nodes require the openclaw CLI to be installed"); - } - - const args = ["agent", ...this.buildAgentArgs(target), "--message", message]; - - // Merge flow-level env vars (state.env) into the child process environment. - // Set CLAWFLOW_NO_SERVE to prevent the child from binding the webhook port. - const env = { - ...process.env, - ...((state.env as Record) ?? {}), - CLAWFLOW_NO_SERVE: "1", - }; - - try { - const { stdout } = await execFileAsync("openclaw", args, { - timeout: nodeTimeoutMs ?? this.cfg.maxNodeDurationMs ?? 120_000, - maxBuffer: 10 * 1024 * 1024, // 10MB - env, - }); - return stdout.trim(); - } catch (err) { - const msg = err instanceof Error ? err.message : String(err); - throw new Error(`openclaw agent failed: ${msg}`); - } + return { output: this.autoParseJson(result) }; } private autoParseJson(value: unknown): unknown { diff --git a/src/core/types.ts b/src/core/types.ts index 01a8acf..9dc22d7 100644 --- a/src/core/types.ts +++ b/src/core/types.ts @@ -512,6 +512,15 @@ export interface ServeConfig { flowsDir?: string; // directory containing .json flow files, default workspace/flows } +export interface ManageConfig { + /** Disable the management API entirely. Default: enabled when a token exists. */ + enabled?: boolean; + /** Port to bind on 127.0.0.1. Default: 18794. */ + port?: number; + /** Bearer token required on every request. Fallback: env CLAWFLOW_ADMIN_TOKEN. */ + token?: string; +} + export interface PluginConfig { apiKey?: string; defaultModel?: string; @@ -535,8 +544,20 @@ export interface PluginConfig { inferenceFn?: InferenceFn; /** OpenClaw agent ID for do:agent nodes (e.g. "ops"). Falls back to --local if unset. */ defaultAgent?: string; + /** + * How `agent` nodes reach a real agent runtime. Defaults to the OpenClaw CLI + * (`OpenClawCliInvoker`); other harnesses supply their own implementation. + */ + agentInvoker?: import("./agent-invoker.js").AgentInvoker; /** Optional HTTP server config — exposes a generic run endpoint per flow. */ serve?: ServeConfig; + /** + * Optional loopback-only management API — exposes the flow_* operations over + * HTTP so out-of-process harness shims (e.g. a Hermes plugin) can register + * the same tool surface and forward calls. Requires a bearer token; never + * starts without one. + */ + manage?: ManageConfig; /** * Optional custom step registry. Defaults to the module-level singleton * populated by `registerStepType()`. Provide a private registry for diff --git a/src/index.ts b/src/index.ts index e685d54..c6021d4 100644 --- a/src/index.ts +++ b/src/index.ts @@ -57,4 +57,11 @@ export type { export { parseDuration, MODEL_MAP, DEFAULT_MODEL, NODE_KEYS } from "./core/types.js"; export { startFlowServer } from "./core/serve.js"; export type { FlowServerOpts } from "./core/serve.js"; -export type { ServeConfig } from "./core/types.js"; +export type { ServeConfig, ManageConfig } from "./core/types.js"; +export { FlowOps } from "./core/ops.js"; +export type { OpResult, FlowOpSpec } from "./core/ops.js"; +export { FLOW_OP_SPECS } from "./core/op-specs.js"; +export { startManagementServer, DEFAULT_MANAGE_PORT } from "./core/manage.js"; +export type { ManageServerOpts } from "./core/manage.js"; +export { OpenClawCliInvoker } from "./core/agent-invoker.js"; +export type { AgentInvoker, AgentTarget, AgentInvokeOptions } from "./core/agent-invoker.js"; diff --git a/src/plugin/index.ts b/src/plugin/index.ts index e206330..2cd8215 100644 --- a/src/plugin/index.ts +++ b/src/plugin/index.ts @@ -1,26 +1,15 @@ -import * as fs from "node:fs"; -import * as path from "node:path"; - -import { FlowRunner, sendEvent } from "../core/runner.js"; - +import { FlowRunner } from "../core/runner.js"; +import { FlowOps } from "../core/ops.js"; +import { FLOW_OP_SPECS } from "../core/op-specs.js"; import { startFlowServer } from "../core/serve.js"; -import { validateFlow } from "../core/validate.js"; -import type { FlowDefinition, FlowNode, PluginConfig, BranchNode, ConditionNode, LoopNode, ParallelNode } from "../core/types.js"; +import { startManagementServer } from "../core/manage.js"; +import type { PluginConfig } from "../core/types.js"; // ---- OpenClaw Plugin: clawflow --------------------------------------------------- -// Registers eleven tools: -// -// flow_create — create a new flow definition and save to file -// flow_delete — soft-delete a flow (moves to .bin/) -// flow_restore_from_bin — restore a flow from the bin or list bin contents -// flow_run — execute a flow (inline or from file) -// flow_resume — resume after approval gate -// flow_send_event — push an event into a waiting flow -// flow_status — inspect a running/completed flow instance -// flow_list — list all saved flow definitions in the workspace -// flow_read — read a flow definition and show expected inputs -// flow_publish — publish a draft flow as a numbered version -// flow_edit — edit nodes in a flow definition (file or inline) +// Thin OpenClaw adapter over the harness-neutral core. Registers one tool per +// FLOW_OP_SPECS entry (flow_create, flow_run, flow_edit, …) forwarding to +// FlowOps, plus the OpenClaw-specific approval gate, the webhook flow server, +// and the loopback management API. interface BeforeToolCallEvent { toolName?: string; @@ -88,7 +77,7 @@ function register(api: PluginApi) { api.logger?.info(`clawflow workspace: ${workspace}`); const runner = new FlowRunner(pluginCfg); - const store = runner.getStore(); + const ops = new FlowOps(workspace, runner); // ---- Flow server (optional) --------------------------------------------------- // Skip when spawned as a child agent (CLAWFLOW_NO_SERVE) to avoid port conflicts. @@ -100,6 +89,24 @@ function register(api: PluginApi) { }); } + // ---- Management API (optional) ------------------------------------------------ + // Loopback-only op surface for out-of-process shims. Never starts without a + // token. Shares CLAWFLOW_NO_SERVE so child agents don't bind ports. + const manageCfg = pluginCfg.manage ?? {}; + const manageToken = manageCfg.token ?? process.env.CLAWFLOW_ADMIN_TOKEN; + if (manageCfg.enabled !== false && manageToken && !process.env.CLAWFLOW_NO_SERVE) { + startManagementServer({ + ops, + token: manageToken, + port: manageCfg.port, + logger: api.logger, + }); + } else if (manageCfg.enabled && !manageToken) { + api.logger?.warn( + "clawflow: manage.enabled is set but no token configured (manage.token or CLAWFLOW_ADMIN_TOKEN) — management API not started", + ); + } + // ---- Approval gate for flow_run ----------------------------------------------- // Flows can call HTTP, exec, and agent tools, so by default we prompt the user // before each run. Disable entirely (`approval.enabled: false`) or skip for @@ -184,1784 +191,28 @@ function register(api: PluginApi) { ); } - // ---- Shared helpers ------------------------------------------------------------ - - /** 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); - } - - /** 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); - } - - /** 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; - } - - /** 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 }; - } - - // ---- flow_create -------------------------------------------------------------- - - api.registerTool( - { - name: "flow_create", - description: `Create a new clawflow definition and save it to a JSON file. - -Builds a FlowDefinition from the provided parameters, validates it, and writes -it to disk. Use this to scaffold a new flow file; use flow_edit to modify it -afterwards and flow_run to execute it. - -Node types: - ai, agent, branch, condition, loop, parallel, http, memory, wait, sleep, code, exec - -All nodes require "name" and "do". Templates: {{ outputKey.field }}.`, - - parameters: { - type: "object", - required: ["file", "flow", "nodes"], - properties: { - file: { - type: "string", - description: - "Filename or path for the new flow file. Plain names like \"my-flow\" are saved to workspace/flows/my-flow.json. Paths with slashes are resolved relative to the workspace.", - }, - flow: { - type: "string", - description: "Unique flow name", - }, - description: { - type: "string", - description: "Human-readable description of what the flow does", - }, - nodes: { - type: "array", - items: { type: "object", additionalProperties: true }, - description: "Array of node definitions", - }, - inputs: { - type: "object", - additionalProperties: true, - description: - 'Declared inputs the flow expects. Map of name → { type?, required?, description?, default? }. Optional: when omitted, the flow accepts any payload. When present, required inputs are enforced before any node runs.', - }, - env: { - type: "object", - additionalProperties: true, - description: "Environment variable defaults", - }, - version: { - type: "string", - description: 'Semver version string, e.g. "1.0.0"', - }, - }, - }, - - async execute( - _id: string, - params: { - file: string; - flow: string; - description?: string; - nodes: FlowNode[]; - inputs?: FlowDefinition["inputs"]; - env?: Record; - version?: string; - }, - ) { - const fs = await import("fs"); - const path = await import("path"); - - const base = workspace; - - let abs: string; - if (params.file.startsWith("/")) { - abs = params.file; - } else if (params.file.includes("/")) { - // Relative path with directory — resolve from workspace root - abs = path.join(base, params.file); - } else { - // Plain name — put in workspace/flows/ - const name = params.file.replace(/\.json$/, ""); - abs = path.join(base, "flows", `${name}.json`); - } - - if (fs.existsSync(abs)) - return { - content: [ - { - type: "text", - text: `File already exists: ${abs}. Use flow_edit to modify it.`, - }, - ], - }; - - const flowDef: FlowDefinition = { - flow: params.flow, - ...(params.version && { version: params.version }), - ...(params.description && { description: params.description }), - ...(params.inputs && { inputs: params.inputs }), - ...(params.env && { env: params.env }), - nodes: params.nodes, - }; - - const validation = validateFlow(flowDef); - if (!validation.ok) - return { - content: [ - { - type: "text", - text: `Validation failed:\n${validation.errors.map((e) => ` - ${e.message}`).join("\n")}`, - }, - ], - }; - - try { - fs.mkdirSync(path.dirname(abs), { recursive: true }); - } catch (err) { - const msg = err instanceof Error ? err.message : String(err); - return { - content: [ - { - type: "text", - text: `Cannot create directory for ${abs}: ${msg}. Is OPENCLAW_WORKSPACE set? Current workspace root: ${base}`, - }, - ], - }; - } - fs.writeFileSync(abs, JSON.stringify(flowDef, null, 2) + "\n"); - - return { - content: [ - { - type: "text", - text: `Flow "${params.flow}" created at ${abs}`, - }, - ], - details: flowDef, - }; - }, - }, - { optional: true }, - ); - - // ---- flow_delete -------------------------------------------------------------- - - api.registerTool( - { - name: "flow_delete", - description: `Delete a flow file by moving it to the bin. - -The flow is not permanently removed — it is timestamped and moved to -workspace/.clawflow/bin/ so it can be restored later with flow_restore_from_bin. -Safe for agents to call without fear of data loss.`, - - parameters: { - type: "object", - required: ["file"], - properties: { - file: { - type: "string", - description: - "Filename or path of the flow to delete. Plain names like \"my-flow\" resolve to workspace/flows/my-flow.json.", - }, - }, - }, - - async execute( - _id: string, - params: { file: string }, - ) { - const fs = await import("fs"); - const path = await import("path"); - - const base = workspace; - - let abs: string; - if (params.file.startsWith("/")) { - abs = params.file; - } else if (params.file.includes("/")) { - abs = path.join(base, params.file); - } else { - const name = params.file.replace(/\.json$/, ""); - abs = path.join(base, "flows", `${name}.json`); - } - - if (!fs.existsSync(abs)) - return { - content: [{ type: "text", text: `File not found: ${abs}` }], - }; - - const binDir = path.join(base, ".clawflow", "bin"); - fs.mkdirSync(binDir, { recursive: true }); - - const basename = path.basename(abs, ".json"); - const ts = new Date().toISOString().replace(/[:.]/g, "-"); - const binPath = path.join(binDir, `${basename}.${ts}.json`); - - fs.renameSync(abs, binPath); - - return { - content: [ - { - type: "text", - text: `Flow moved to bin: ${binPath}`, - }, - ], - }; - }, - }, - { optional: true }, - ); - - // ---- flow_restore_from_bin ------------------------------------------------------------- - - api.registerTool( - { - name: "flow_restore_from_bin", - description: `Restore a flow from the bin or list bin contents. - -Without "name", lists all flows in workspace/.clawflow/bin/ with their timestamps. -With "name", restores the most recent version of that flow back to the flows/ -directory. If the flow file already exists, the restore is rejected.`, - - parameters: { - type: "object", - properties: { - name: { - type: "string", - description: - "Flow name to restore (without timestamp). Omit to list all bin contents.", - }, - }, - }, - - async execute( - _id: string, - params: { name?: string }, - ) { - const fs = await import("fs"); - const path = await import("path"); - - const base = workspace; - const binDir = path.join(base, ".clawflow", "bin"); - - if (!fs.existsSync(binDir)) - return { - content: [{ type: "text", text: "Bin is empty." }], - }; - - const files = fs.readdirSync(binDir) - .filter((f: string) => f.endsWith(".json")) - .sort() - .reverse(); - - if (files.length === 0) - return { - content: [{ type: "text", text: "Bin is empty." }], - }; - - // ---- List mode --- - if (!params.name) { - const entries = files.map((f: string) => { - const match = f.match(/^(.+?)\.(\d{4}-.+)\.json$/); - return { - name: match?.[1] ?? f, - deletedAt: match?.[2] ?? "unknown", - file: f, - }; - }); - return { - content: [ - { type: "text", text: JSON.stringify(entries, null, 2) }, - ], - details: entries, - }; - } - - // ---- Restore mode --- - const prefix = params.name.replace(/\.json$/, ""); - const match = files.find((f: string) => f.startsWith(`${prefix}.`)); - - if (!match) - return { - content: [ - { - type: "text", - text: `No bin entry found for "${prefix}".`, - }, - ], - }; - - const dest = path.join(base, "flows", `${prefix}.json`); - if (fs.existsSync(dest)) - return { - content: [ - { - type: "text", - text: `Cannot restore: ${dest} already exists. Delete or rename it first.`, - }, - ], - }; - - fs.renameSync(path.join(binDir, match), dest); - - return { - content: [ - { - type: "text", - text: `Restored "${prefix}" from bin to ${dest}`, - }, - ], - }; - }, - }, - { optional: true }, - ); - - // ---- flow_run ----------------------------------------------------------------- - - api.registerTool( - { - name: "flow_run", - catalogMode: "direct-only", - description: `Run an agentic workflow in the clawflow format. - -State model: - The "input" parameter becomes "inputs" in flow state (i.e. state.inputs). - Flow state = { inputs, env?, ...nodeOutputs }. - Each node with "output" adds its result to state (e.g. output: "result" → state.result). - In code nodes: fn(input, state) — "input" is the resolved node.input field, "state" is the full flow state. - IMPORTANT: inputs contains ALL initial data. If you need different parts of inputs in a code node, - use object-style input: { "payload": "inputs.payload", "email": "inputs.email_to" } - or access via state.inputs.field inside the code. - -Node types: - ai — LLM call, structured or freeform. Use schema: for typed output. - agent — delegate to a real OpenClaw agent. Fields mirror the CLI flags: agent - (a configured slug, e.g. "clawflow" → --agent), sessionKey (a session - key, e.g. "agent:main:slack:channel:agent" → --session-key), sessionId - (→ --session-id), channel (→ --channel). A bare session key is scoped by - agent; an "agent:"-prefixed key is self-scoping. (agentId/session are - deprecated aliases for agent/sessionKey.) - branch — route to different nodes based on a value: { on, paths, default } - loop — iterate over a list: { over, as, nodes[] } - parallel — run nodes concurrently: { nodes[], mode: "all"|"race" } - http — call an external API: { url, method, body, headers } - memory — persist data: { action: read|write|delete, key, value } - wait — pause for approval or event: { for: "approval"|"event", event?, timeout? } - sleep — pause for duration: { duration: "5m" } - code — inline JS expression: { run: "...", input? } + // ---- Tools -------------------------------------------------------------------- + // One tool per op spec, forwarding to FlowOps and adapting the { text, + // details? } result to OpenClaw's tool-result shape. -All nodes support retry: { limit, delay, backoff } and timeout. -Templates: use {{ nodeName.field }} to reference any value in flow state. -Returns instanceId for status tracking and resume. - -Versioning: when a flow has published versions, flow_run uses the latest -published version by default. Set draft: true to run the working copy instead. -Set version to run a specific published version.`, - - parameters: { - type: "object", - properties: { - flow: { - type: "object", - properties: { - flow: { type: "string" }, - description: { type: "string" }, - nodes: { - type: "array", - items: { type: "object", additionalProperties: true }, - }, - }, - required: ["flow", "nodes"], - additionalProperties: true, - }, - file: { - type: "string", - description: "Path to a .json flow file (or plain name like \"my-flow\")", - }, - input: { - type: "object", - additionalProperties: true, - description: "Input data, available as inputs.* in the flow (and at state.inputs in code nodes).", - }, - draft: { - type: "boolean", - description: - "Run the draft (working copy) instead of the latest published version. Default: false.", - }, - version: { - type: "number", - description: - "Run a specific published version number. Overrides draft flag.", - }, - }, - }, - - async execute( - _id: string, - params: { - flow?: FlowDefinition; - file?: string; - input?: unknown; - draft?: boolean; - version?: number; - }, - ) { - let flowDef: FlowDefinition; - let source = ""; - - if (params.file) { - const { readFileSync, existsSync } = await import("fs"); - const pathMod = await import("path"); - const abs = resolveFlowFile(params.file); - - // Determine flow name for version lookup - const flowName = pathMod.basename(abs, ".json"); - - if (params.version != null) { - // Specific version requested - const vDef = readVersion(flowName, params.version); - if (!vDef) - return { - content: [ - { - type: "text", - text: `Version ${params.version} not found for flow "${flowName}". Available: ${listVersions(flowName).join(", ") || "none (not yet published)"}`, - }, - ], - }; - flowDef = vDef; - source = `v${params.version}`; - } else if (!params.draft) { - // Default: use latest published version if available - const latest = readLatestVersion(flowName); - if (latest) { - flowDef = latest.def; - source = `v${latest.version}`; - } else { - // No published versions — fall back to draft - if (!existsSync(abs)) - return { - content: [{ type: "text", text: `File not found: ${abs}` }], - }; - flowDef = JSON.parse(readFileSync(abs, "utf8")) as FlowDefinition; - source = "draft (no published versions)"; - } - } else { - // Explicit draft mode - if (!existsSync(abs)) - return { - content: [{ type: "text", text: `File not found: ${abs}` }], - }; - flowDef = JSON.parse(readFileSync(abs, "utf8")) as FlowDefinition; - source = "draft"; - } - } else if (params.flow) { - flowDef = params.flow as FlowDefinition; - source = "inline"; - } else { - return { - content: [ - { - type: "text", - text: "Provide either `flow` (inline definition) or `file` (path).", - }, - ], - }; - } - - try { - const result = await runner.run(flowDef, params.input ?? {}); - const out = { ...result, _source: source }; - return { - content: [ - { type: "text", text: JSON.stringify(out, null, 2) }, - ], - details: out, - }; - } catch (err) { - return { - content: [ - { - type: "text", - text: `Error: ${err instanceof Error ? err.message : String(err)}`, - }, - ], - }; - } - }, - }, - { optional: true }, - ); - - // ---- flow_resume -------------------------------------------------------------- - - api.registerTool( - { - name: "flow_resume", - description: `Resume a paused clawflow after an approval gate. -Use the instanceId (= resumeToken) from a flow_run result where status was "paused". -Set approved=true to continue, false to cancel. -You must pass the original flow definition back so the runner can continue.`, - - parameters: { - type: "object", - required: ["instanceId", "approved", "flow"], - properties: { - instanceId: { - type: "string", - description: "The instanceId from the paused flow_run result", - }, - approved: { type: "boolean" }, - flow: { - type: "object", - required: ["flow", "nodes"], - properties: { - flow: { type: "string" }, - nodes: { - type: "array", - items: { type: "object", additionalProperties: true }, - }, - }, - additionalProperties: true, - }, - }, - }, - - async execute( - _id: string, - params: { - instanceId: string; - approved: boolean; - flow: FlowDefinition; - }, - ) { - try { - const result = await runner.resume( - params.instanceId, - params.flow, - params.approved, - ); - return { - content: [ - { type: "text", text: JSON.stringify(result, null, 2) }, - ], - details: result, - }; - } catch (err) { - return { - content: [ - { - type: "text", - text: `Error: ${err instanceof Error ? err.message : String(err)}`, - }, - ], - }; - } - }, - }, - { optional: true }, - ); - - // ---- flow_send_event ---------------------------------------------------------- - - api.registerTool( - { - name: "flow_send_event", - description: `Send an event to a flow instance that is waiting with do: wait / for: event. -This is the equivalent of Cloudflare's instance.sendEvent(). -The eventType must match the "event" field on the wait node. -payload is passed as the output of the wait node and into flow state.`, - - parameters: { - type: "object", - required: ["instanceId", "eventType"], - properties: { - instanceId: { type: "string" }, - eventType: { - type: "string", - description: "Must match the 'event' field of the wait node", - }, - payload: { - type: "object", - additionalProperties: true, - }, - }, - }, - - async execute( - _id: string, - params: { - instanceId: string; - eventType: string; - payload?: unknown; - }, - ) { - const delivered = sendEvent( - params.instanceId, - params.eventType, - params.payload ?? {}, - ); - if (!delivered) { + for (const spec of FLOW_OP_SPECS) { + api.registerTool( + { + name: spec.name, + ...(spec.catalogMode && { catalogMode: spec.catalogMode }), + description: spec.description, + parameters: spec.parameters, + async execute(_id: string, params: unknown) { + const result = await ops.execute(spec.name, params); return { - content: [ - { - type: "text", - text: `No active waiter found for instance "${params.instanceId}" event "${params.eventType}". The flow may not be waiting, or the event type doesn't match.`, - }, - ], + content: [{ type: "text", text: result.text }], + ...(result.details !== undefined && { details: result.details }), }; - } - return { - content: [ - { - type: "text", - text: `Event "${params.eventType}" delivered to instance "${params.instanceId}".`, - }, - ], - }; - }, - }, - { optional: true }, - ); - - // ---- flow_status -------------------------------------------------------------- - - api.registerTool( - { - name: "flow_status", - description: `Get the status and state of a flow instance, or list all instances. -Status values: running | completed | paused | waiting | failed | cancelled`, - - parameters: { - type: "object", - properties: { - instanceId: { - type: "string", - description: "Specific instance to inspect. Omit to list all.", - }, - filter: { - type: "string", - description: - "Filter by status: running | completed | paused | waiting | failed | cancelled", - }, }, }, - - async execute( - _id: string, - params: { instanceId?: string; filter?: string }, - ) { - if (params.instanceId) { - const record = store.get(params.instanceId); - if (!record) - return { - content: [ - { - type: "text", - text: `Instance not found: ${params.instanceId}`, - }, - ], - }; - return { - content: [ - { type: "text", text: JSON.stringify(record, null, 2) }, - ], - details: record, - }; - } - const records = store.list(params.filter); - const summary = records.map((r) => ({ - instanceId: r.instanceId, - flow: r.flowName, - status: r.status, - updatedAt: r.updatedAt, - waitingFor: r.waitingFor, - })); - return { - content: [ - { type: "text", text: JSON.stringify(summary, null, 2) }, - ], - details: summary, - }; - }, - }, - { optional: true }, - ); - - // ---- flow_list ---------------------------------------------------------------- - - api.registerTool( - { - name: "flow_list", - catalogMode: "direct-only", - description: `List all saved flow definitions in the workspace. - -Scans the flows directory for .json files and returns a summary of each flow -including its name, description, declared inputs, version, node count, and file path. -Use this to discover available flows before running or editing them.`, - - parameters: { - type: "object", - properties: { - dir: { - type: "string", - description: - "Directory to scan. Defaults to workspace/flows/. Absolute paths are used as-is; relative paths resolve from the workspace root.", - }, - }, - }, - - async execute( - _id: string, - params: { dir?: string }, - ) { - const fs = await import("fs"); - const path = await import("path"); - - const base = workspace; - - let dir: string; - if (!params.dir) { - dir = path.join(base, "flows"); - } else if (params.dir.startsWith("/")) { - dir = params.dir; - } else { - dir = path.join(base, params.dir); - } - - if (!fs.existsSync(dir)) { - return { - content: [ - { - type: "text", - text: `Flows directory not found: ${dir}`, - }, - ], - }; - } - - const files = fs.readdirSync(dir).filter((f: string) => f.endsWith(".json")); - - if (files.length === 0) { - return { - content: [ - { - type: "text", - text: `No flow files found in ${dir}`, - }, - ], - }; - } - - const flows: Array<{ - file: string; - flow: string; - description?: string; - inputs?: FlowDefinition["inputs"]; - expectedInputs?: string[]; - publishedVersion?: number; - totalVersions?: number; - nodes: number; - }> = []; - - for (const file of files) { - const abs = path.join(dir, file); - try { - const raw = fs.readFileSync(abs, "utf-8"); - const def = JSON.parse(raw) as FlowDefinition; - if (!def.flow || !Array.isArray(def.nodes)) continue; - const referenced = extractInputRefs(def.nodes); - const flowName = file.replace(/\.json$/, ""); - const versions = listVersions(flowName); - flows.push({ - file: abs, - flow: def.flow, - ...(def.description && { description: def.description }), - ...(def.inputs && { inputs: def.inputs }), - ...(referenced.length > 0 && { expectedInputs: referenced }), - ...(versions.length > 0 && { - publishedVersion: versions[versions.length - 1], - totalVersions: versions.length, - }), - nodes: def.nodes.length, - }); - } catch { - // skip non-flow JSON files - } - } - - if (flows.length === 0) { - return { - content: [ - { - type: "text", - text: `No valid flow definitions found in ${dir}`, - }, - ], - }; - } - - return { - content: [ - { type: "text", text: JSON.stringify(flows, null, 2) }, - ], - details: flows, - }; - }, - }, - { optional: true }, - ); - - // ---- flow_read ---------------------------------------------------------------- - - /** - * Recursively extract unique inputs.* paths from any string values in an - * object tree. This tells agents what input fields a flow references — used - * when the flow has no declared inputs block to give a best-effort hint. - */ - function extractInputRefs(obj: unknown): string[] { - const paths = new Set(); - const re = /\{\{\s*inputs\.(\w+(?:\.\w+)*)/g; - - function walk(val: unknown): void { - if (typeof val === "string") { - let m: RegExpExecArray | null; - while ((m = re.exec(val)) !== null) paths.add(m[1]); - } else if (Array.isArray(val)) { - for (const item of val) walk(item); - } else if (val && typeof val === "object") { - for (const v of Object.values(val)) walk(v); - } - } - walk(obj); - return [...paths].sort(); - } - - /** Find a node by name, searching nested structures (branch paths, loop, parallel, condition). */ - function findNode(nodes: FlowNode[], name: string): FlowNode | undefined { - for (const n of nodes) { - if (n.name === name) return n; - if ("paths" in n && n.paths) { - for (const branch of Object.values( - n.paths as Record, - )) { - const found = findNode(branch, name); - if (found) return found; - } - } - if ("default" in n && Array.isArray(n.default)) { - const found = findNode(n.default as FlowNode[], name); - if (found) return found; - } - if ("nodes" in n && Array.isArray(n.nodes)) { - const found = findNode(n.nodes as FlowNode[], name); - if (found) return found; - } - if ("then" in n && Array.isArray(n.then)) { - const found = findNode(n.then as FlowNode[], name); - if (found) return found; - } - if ("else" in n && Array.isArray((n as ConditionNode).else)) { - const found = findNode((n as ConditionNode).else!, name); - if (found) return found; - } - } - return undefined; + { optional: true }, + ); } - - api.registerTool( - { - name: "flow_read", - description: `Read a flow definition from file and return its contents. - -Returns the full flow definition (or a single node if specified). The response -includes the declared "inputs" block when present, plus a best-effort list of -input fields referenced by templates (extracted from {{ inputs.* }} usages). -Use this to inspect a flow before running it or to understand what inputs it -needs. - -Versioning: by default reads the draft (working copy). Set version to read -a specific published version. The response includes available version numbers.`, - - parameters: { - type: "object", - required: ["file"], - properties: { - file: { - type: "string", - description: - "Filename or path to the flow file. Plain names resolve to workspace/flows/.json.", - }, - node: { - type: "string", - description: - "Name of a specific node to return. Searches nested structures (branches, loops, etc.).", - }, - version: { - type: "number", - description: - "Read a specific published version instead of the draft.", - }, - }, - }, - - async execute( - _id: string, - params: { file: string; node?: string; version?: number }, - ) { - const fs = await import("fs"); - const pathMod = await import("path"); - - const abs = resolveFlowFile(params.file); - const flowName = pathMod.basename(abs, ".json"); - const versions = listVersions(flowName); - - let flowDef: FlowDefinition; - let source: string; - - if (params.version != null) { - const vDef = readVersion(flowName, params.version); - if (!vDef) { - return { - content: [ - { - type: "text", - text: `Version ${params.version} not found for "${flowName}". Available: ${versions.join(", ") || "none"}`, - }, - ], - }; - } - flowDef = vDef; - source = `v${params.version}`; - } else { - if (!fs.existsSync(abs)) { - return { - content: [{ type: "text", text: `File not found: ${abs}` }], - }; - } - try { - flowDef = JSON.parse(fs.readFileSync(abs, "utf-8")) as FlowDefinition; - } catch (err) { - return { - content: [ - { - type: "text", - text: `Failed to parse ${abs}: ${err instanceof Error ? err.message : String(err)}`, - }, - ], - }; - } - source = "draft"; - } - - const referenced = extractInputRefs(flowDef.nodes); - - // Single node mode - if (params.node) { - const found = findNode(flowDef.nodes, params.node); - if (!found) { - return { - content: [ - { - type: "text", - text: `Node "${params.node}" not found in flow "${flowDef.flow}". Nodes: ${flowDef.nodes.map((n) => n.name).join(", ")}`, - }, - ], - }; - } - const nodeInputs = extractInputRefs(found); - const result = { - flow: flowDef.flow, - _source: source, - _file: abs, - ...(versions.length > 0 && { _versions: versions }), - ...(nodeInputs.length > 0 && { expectedInputs: nodeInputs }), - node: found, - }; - return { - content: [{ type: "text", text: JSON.stringify(result, null, 2) }], - details: result, - }; - } - - // Full flow mode - const result = { - ...flowDef, - _source: source, - _file: abs, - ...(versions.length > 0 && { _versions: versions }), - ...(referenced.length > 0 && { _expectedInputs: referenced }), - }; - return { - content: [{ type: "text", text: JSON.stringify(result, null, 2) }], - details: result, - }; - }, - }, - { optional: true }, - ); - - // ---- flow_publish -------------------------------------------------------------- - - api.registerTool( - { - name: "flow_publish", - description: `Publish the current draft of a flow as a new numbered version. - -Validates the draft, assigns the next version number (auto-incrementing integer), -and saves an immutable copy to .clawflow/versions//.json. -After publishing, flow_run will use this version by default. - -Use this when a flow is ready for production. Edits via flow_edit continue to -modify the draft without affecting published versions.`, - - parameters: { - type: "object", - required: ["file"], - properties: { - file: { - type: "string", - description: - "Filename or path to the draft flow file. Plain names resolve to workspace/flows/.json.", - }, - }, - }, - - async execute( - _id: string, - params: { file: string }, - ) { - const fs = await import("fs"); - const pathMod = await import("path"); - - const abs = resolveFlowFile(params.file); - if (!fs.existsSync(abs)) { - return { - content: [{ type: "text", text: `Draft not found: ${abs}` }], - }; - } - - let flowDef: FlowDefinition; - try { - flowDef = JSON.parse(fs.readFileSync(abs, "utf-8")) as FlowDefinition; - } catch (err) { - return { - content: [ - { - type: "text", - text: `Failed to parse ${abs}: ${err instanceof Error ? err.message : String(err)}`, - }, - ], - }; - } - - const validation = validateFlow(flowDef); - if (!validation.ok) { - return { - content: [ - { - type: "text", - text: `Validation failed — cannot publish:\n${validation.errors.map((e) => ` - ${e.message}`).join("\n")}`, - }, - ], - }; - } - - 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"); - - return { - content: [ - { - type: "text", - text: `Published "${flowDef.flow}" as v${nextVersion}. flow_run will now use this version by default.\nFile: ${versionFile}`, - }, - ], - details: { - flow: flowDef.flow, - version: nextVersion, - file: versionFile, - totalVersions: nextVersion, - }, - }; - }, - }, - { optional: true }, - ); - - // ---- flow_edit ---------------------------------------------------------------- - - api.registerTool( - { - name: "flow_edit", - description: `Edit nodes in a clawflow definition. Operates on a file or inline flow. - -Actions: - set — set top-level flow fields (description, inputs, env, version) - update — update a node entirely or patch specific fields by node name - add — insert a new node at a position (default: end) - remove — remove a node by name - move — move a node to a new position (same level or into a parent) - wrap — wrap one or more nodes into a new container (loop, condition, branch, parallel) - revert — undo the last edit (restores the previous version from history) - list — list all nodes with index, name, type, and output key - -All actions that target nodes (update, remove, move, add) search recursively -through nested structures (branch paths, condition then/else, loop, parallel). - -The "parent" parameter targets a nested node list using slash-separated paths: - "myBranch/true" → branch "myBranch", path "true" - "myCond/then" → condition "myCond", then block - "myLoop" → loop "myLoop", child nodes - "outer/true/inner" → chained nesting - -The edited flow is validated after every mutation. If validation fails, the -edit is rejected and errors are returned. For file-based flows, the file is -overwritten with the updated definition on success. - -Examples: - Set flow fields: { action: "set", fields: { description: "New desc", inputs: { ticket_id: { type: "string", required: true } } } } - Update one field: { action: "update", node: "classify", fields: { prompt: "New prompt" } } - Replace full node: { action: "update", node: "classify", replace: { name: "classify", do: "ai", prompt: "..." } } - Add at position: { action: "add", position: 2, nodeDefinition: { name: "step3", do: "code", run: "..." } } - Add inside branch: { action: "add", parent: "shouldUpdate/true", nodeDefinition: { name: "step3", do: "code", run: "..." } } - Remove: { action: "remove", node: "old-step" } - Move into loop: { action: "move", node: "step3", parent: "myLoop", position: 0 } - Move into else: { action: "move", node: "step3", parent: "myCond/else", position: 0 } - Wrap in loop: { action: "wrap", nodes: ["step1", "step2"], wrapper: { name: "myLoop", do: "loop", over: "items", as: "item" } } - Wrap in condition: { action: "wrap", nodes: ["step1", "step2"], wrapper: { name: "guard", do: "condition", if: "{{ items.length > 0 }}" } } - List: { action: "list" } - -To RESTRUCTURE an existing flow — make a group of nodes conditional, nest them in a -loop, or reparent them — use "wrap" (many nodes at once) or "move" (one node into a -"parent" path like "guard/else"). Do NOT remove-and-re-add each node, and do NOT -re-send the whole flow: a single large edit is slow and can stall mid-generation.`, - - parameters: { - type: "object", - required: ["action"], - properties: { - file: { - type: "string", - description: "Path to a .json flow file. Mutually exclusive with 'flow'.", - }, - flow: { - type: "object", - description: "Inline flow definition. Mutually exclusive with 'file'.", - properties: { - flow: { type: "string" }, - nodes: { - type: "array", - items: { type: "object", additionalProperties: true }, - }, - }, - required: ["flow", "nodes"], - additionalProperties: true, - }, - action: { - type: "string", - enum: ["set", "update", "add", "remove", "move", "wrap", "revert", "list"], - }, - node: { - type: "string", - description: "Node name to target (required for update, remove, move). Searched recursively through nested structures.", - }, - fields: { - type: "object", - additionalProperties: true, - description: "For action=set: top-level flow fields to set (description, inputs, env, version). For action=update: partial field updates to merge into the node.", - }, - replace: { - type: "object", - additionalProperties: true, - description: "For action=update: full node replacement (must include name and do)", - }, - nodeDefinition: { - type: "object", - additionalProperties: true, - description: "For action=add: the new node definition", - }, - position: { - type: "number", - description: "For action=add/move: index to insert/move to (0-based). Default: end.", - }, - parent: { - type: "string", - description: 'For action=add/move: target a nested node list. Slash-separated path e.g. "myBranch/true", "myCond/then", "myLoop". For move: destination parent (node is removed from current location and inserted here).', - }, - nodes: { - type: "array", - items: { type: "string" }, - description: "For action=wrap: array of node names to wrap into a container.", - }, - wrapper: { - type: "object", - additionalProperties: true, - description: "For action=wrap: the container node definition (must include name, do). Wrapped nodes become its children (e.g. loop.nodes, condition.then, branch.paths.true, parallel.nodes).", - }, - }, - }, - - async execute( - _id: string, - params: { - file?: string; - flow?: FlowDefinition; - action: "set" | "update" | "add" | "remove" | "move" | "wrap" | "revert" | "list"; - node?: string; - fields?: Record; - replace?: FlowNode; - nodeDefinition?: FlowNode; - position?: number; - parent?: string; - nodes?: string[]; - wrapper?: FlowNode; - }, - ) { - // ---- Load flow definition --- - let flowDef: FlowDefinition; - let filePath: string | undefined; - - if (params.file) { - const fs = await import("fs"); - const pathMod = await import("path"); - const base = workspace; - let abs: string; - if (params.file.startsWith("/")) { - abs = params.file; - } else if (params.file.includes("/")) { - abs = pathMod.join(base, params.file); - } else { - const name = params.file.replace(/\.json$/, ""); - abs = pathMod.join(base, "flows", `${name}.json`); - } - if (!fs.existsSync(abs)) - return { - content: [{ type: "text", text: `File not found: ${abs}` }], - }; - filePath = abs; - flowDef = JSON.parse(fs.readFileSync(abs, "utf8")) as FlowDefinition; - } else if (params.flow) { - flowDef = params.flow; - } else { - return { - content: [ - { - type: "text", - text: "Provide either `file` (path) or `flow` (inline definition).", - }, - ], - }; - } - - const ok = (msg: string, flow: FlowDefinition) => ({ - content: [{ type: "text", text: msg }], - details: flow, - }); - const fail = (msg: string) => ({ - content: [{ type: "text", text: msg }], - }); - - // Resolve a node list from a parent path. Supports: - // undefined → flowDef.nodes (top-level) - // "myBranch/true" → branch node "myBranch", paths["true"] - // "myBranch/default" → branch node "myBranch", default - // "myCond/then" → condition node "myCond", then - // "myCond/else" → condition node "myCond", else - // "myLoop" → loop node "myLoop", nodes - // "myParallel" → parallel node "myParallel", nodes - // Paths can be chained: "outerBranch/true/innerLoop" - const resolveNodeList = ( - parentPath?: string, - ): { nodes: FlowNode[]; error?: string } => { - if (!parentPath) return { nodes: flowDef.nodes }; - - const parts = parentPath.split("/"); - let current: FlowNode[] = flowDef.nodes; - - for (let i = 0; i < parts.length; i++) { - const part = parts[i]; - const node = current.find((n) => n.name === part); - if (!node) - return { - nodes: [], - error: `Parent node "${part}" not found at level ${i}.`, - }; - - if (node.do === "branch") { - const b = node as BranchNode; - const key = parts[++i]; // consume next part as path key - if (!key) - return { - nodes: [], - error: `Branch "${part}" requires a path key (e.g., "${part}/true").`, - }; - if (key === "default") { - if (!b.default) b.default = []; - current = b.default; - } else { - if (!b.paths[key]) b.paths[key] = []; - current = b.paths[key]; - } - } else if (node.do === "condition") { - const c = node as ConditionNode; - const key = parts[++i]; // consume next part as then/else - if (!key) - return { - nodes: [], - error: `Condition "${part}" requires "then" or "else" (e.g., "${part}/then").`, - }; - if (key === "then") { - current = c.then; - } else if (key === "else") { - if (!c.else) c.else = []; - current = c.else; - } else { - return { - nodes: [], - error: `Condition "${part}" only accepts "then" or "else", got "${key}".`, - }; - } - } else if (node.do === "loop") { - current = (node as LoopNode).nodes; - } else if (node.do === "parallel") { - current = (node as ParallelNode).nodes; - } else { - return { - nodes: [], - error: `Node "${part}" (do: "${node.do}") has no child nodes.`, - }; - } - } - return { nodes: current }; - }; - - // Find a node by name, optionally scoped to a parent path - const findIndex = (name: string, parentPath?: string) => { - const { nodes, error } = resolveNodeList(parentPath); - if (error) return -1; - return nodes.findIndex((n) => n.name === name); - }; - - // Deep-find: search all node lists recursively, return the containing array + index - const deepFind = ( - name: string, - nodes: FlowNode[] = flowDef.nodes, - ): { list: FlowNode[]; index: number } | null => { - const idx = nodes.findIndex((n) => n.name === name); - if (idx !== -1) return { list: nodes, index: idx }; - for (const node of nodes) { - const children = getChildLists(node); - for (const child of children) { - const found = deepFind(name, child); - if (found) return found; - } - } - return null; - }; - - // Get all child node arrays from a container node - const getChildLists = (node: FlowNode): FlowNode[][] => { - switch (node.do) { - case "branch": { - const b = node as BranchNode; - const lists = Object.values(b.paths); - if (b.default) lists.push(b.default); - return lists; - } - case "condition": { - const c = node as ConditionNode; - const lists: FlowNode[][] = [c.then]; - if (c.else) lists.push(c.else); - return lists; - } - case "loop": - return [(node as LoopNode).nodes]; - case "parallel": - return [(node as ParallelNode).nodes]; - default: - return []; - } - }; - - // Save a snapshot before mutating (file-based flows only). - // Derive history dir from the flow file's parent (sibling to flows/). - const saveSnapshot = async () => { - if (!filePath) return; - const fs = await import("fs"); - const pathMod = await import("path"); - const flowDir = pathMod.dirname(filePath); - const root = pathMod.dirname(flowDir); // parent of flows/ - const flowName = pathMod.basename(filePath, ".json"); - const histDir = pathMod.join(root, ".clawflow", "history", flowName); - try { - fs.mkdirSync(histDir, { recursive: true }); - const ts = new Date().toISOString().replace(/[:.]/g, "-"); - const snapPath = pathMod.join(histDir, `${ts}.json`); - fs.writeFileSync(snapPath, fs.readFileSync(filePath, "utf8")); - } catch { - // History is best-effort — don't block the edit - } - }; - - // ---- Revert (undo last edit) --- - if (params.action === "revert") { - if (!filePath) - return fail("Revert only works on file-based flows."); - const fs = await import("fs"); - const pathMod = await import("path"); - const flowDir = pathMod.dirname(filePath); - const root = pathMod.dirname(flowDir); - const flowName = pathMod.basename(filePath, ".json"); - const histDir = pathMod.join(root, ".clawflow", "history", flowName); - if (!fs.existsSync(histDir)) - return fail("No edit history found for this flow."); - const snaps = fs.readdirSync(histDir) - .filter((f: string) => f.endsWith(".json")) - .sort() - .reverse(); - if (snaps.length === 0) - return fail("No edit history found for this flow."); - const latest = pathMod.join(histDir, snaps[0]); - const restored = JSON.parse(fs.readFileSync(latest, "utf8")) as FlowDefinition; - fs.writeFileSync(filePath, JSON.stringify(restored, null, 2) + "\n"); - fs.unlinkSync(latest); - return ok( - `Reverted to snapshot ${snaps[0]}. File written: ${filePath}`, - restored, - ); - } - - // ---- Set (top-level flow fields) --- - if (params.action === "set") { - if (!params.fields) - return fail('action "set" requires "fields".'); - - const allowed = ["description", "inputs", "env", "version"]; - const backup = { ...flowDef }; - - for (const [key, value] of Object.entries(params.fields)) { - if (!allowed.includes(key)) - return fail( - `Cannot set "${key}". Allowed fields: ${allowed.join(", ")}. Use "update" to modify nodes.`, - ); - if (value === null || value === undefined) { - delete (flowDef as unknown as Record)[key]; - } else { - (flowDef as unknown as Record)[key] = value; - } - } - - const validation = validateFlow(flowDef); - if (!validation.ok) { - // rollback - Object.assign(flowDef, backup); - return fail( - `Validation failed after set:\n${validation.errors.map((e) => ` - ${e.message}`).join("\n")}`, - ); - } - - if (filePath) { - await saveSnapshot(); - const { writeFileSync } = await import("fs"); - writeFileSync(filePath, JSON.stringify(flowDef, null, 2) + "\n"); - } - return ok( - `Flow fields updated: ${Object.keys(params.fields).join(", ")}.${filePath ? ` File written: ${filePath}` : ""}`, - flowDef, - ); - } - - // ---- List --- - if (params.action === "list") { - const list = flowDef.nodes.map((n, i) => ({ - index: i, - name: n.name, - do: n.do, - output: n.output ?? null, - })); - return { - content: [ - { type: "text", text: JSON.stringify(list, null, 2) }, - ], - details: list, - }; - } - - // ---- Update --- - if (params.action === "update") { - if (!params.node) - return fail('action "update" requires "node" (node name).'); - const found = deepFind(params.node); - if (!found) - return fail(`Node "${params.node}" not found.`); - - if (params.replace) { - found.list[found.index] = params.replace as FlowNode; - } else if (params.fields) { - found.list[found.index] = { - ...found.list[found.index], - ...params.fields, - } as FlowNode; - } else { - return fail( - 'action "update" requires either "fields" (partial) or "replace" (full node).', - ); - } - - const validation = validateFlow(flowDef); - if (!validation.ok) - return fail( - `Validation failed after update:\n${validation.errors.map((e) => ` - ${e.message}`).join("\n")}`, - ); - - if (filePath) { - await saveSnapshot(); - const { writeFileSync } = await import("fs"); - writeFileSync(filePath, JSON.stringify(flowDef, null, 2) + "\n"); - } - return ok( - `Node "${params.node}" updated.${filePath ? ` File written: ${filePath}` : ""}`, - flowDef, - ); - } - - // ---- Add --- - if (params.action === "add") { - if (!params.nodeDefinition) - return fail('action "add" requires "nodeDefinition".'); - const { nodes: targetList, error: parentErr } = resolveNodeList(params.parent); - if (parentErr) return fail(parentErr); - const pos = params.position ?? targetList.length; - if (pos < 0 || pos > targetList.length) - return fail( - `Position ${pos} out of range (0-${targetList.length}).`, - ); - targetList.splice(pos, 0, params.nodeDefinition as FlowNode); - - const validation = validateFlow(flowDef); - if (!validation.ok) { - targetList.splice(pos, 1); // rollback - return fail( - `Validation failed after add:\n${validation.errors.map((e) => ` - ${e.message}`).join("\n")}`, - ); - } - - if (filePath) { - await saveSnapshot(); - const { writeFileSync } = await import("fs"); - writeFileSync(filePath, JSON.stringify(flowDef, null, 2) + "\n"); - } - const loc = params.parent ? ` in ${params.parent}` : ""; - return ok( - `Node "${params.nodeDefinition.name}" added at position ${pos}${loc}.${filePath ? ` File written: ${filePath}` : ""}`, - flowDef, - ); - } - - // ---- Remove --- - if (params.action === "remove") { - if (!params.node) - return fail('action "remove" requires "node" (node name).'); - const found = deepFind(params.node); - if (!found) - return fail(`Node "${params.node}" not found.`); - - const removed = found.list.splice(found.index, 1)[0]; - - const validation = validateFlow(flowDef); - if (!validation.ok) { - found.list.splice(found.index, 0, removed); // rollback - return fail( - `Validation failed after remove (rolled back):\n${validation.errors.map((e) => ` - ${e.message}`).join("\n")}`, - ); - } - - if (filePath) { - await saveSnapshot(); - const { writeFileSync } = await import("fs"); - writeFileSync(filePath, JSON.stringify(flowDef, null, 2) + "\n"); - } - return ok( - `Node "${params.node}" removed.${filePath ? ` File written: ${filePath}` : ""}`, - flowDef, - ); - } - - // ---- Move --- - if (params.action === "move") { - if (!params.node) - return fail('action "move" requires "node" (node name).'); - if (params.position === undefined) - return fail('action "move" requires "position".'); - - // Find the node wherever it is - const source = deepFind(params.node); - if (!source) - return fail(`Node "${params.node}" not found.`); - - // Resolve destination list - const { nodes: destList, error: parentErr } = resolveNodeList(params.parent); - if (parentErr) return fail(parentErr); - - // Remove from source - const [moved] = source.list.splice(source.index, 1); - const pos = Math.min(params.position, destList.length); - destList.splice(pos, 0, moved); - - const validation = validateFlow(flowDef); - if (!validation.ok) { - // rollback: remove from dest, re-insert at source - destList.splice(pos, 1); - source.list.splice(source.index, 0, moved); - return fail( - `Validation failed after move (rolled back):\n${validation.errors.map((e) => ` - ${e.message}`).join("\n")}`, - ); - } - - if (filePath) { - await saveSnapshot(); - const { writeFileSync } = await import("fs"); - writeFileSync(filePath, JSON.stringify(flowDef, null, 2) + "\n"); - } - const loc = params.parent ? ` in ${params.parent}` : ""; - return ok( - `Node "${params.node}" moved to position ${pos}${loc}.${filePath ? ` File written: ${filePath}` : ""}`, - flowDef, - ); - } - - // ---- Wrap --- - if (params.action === "wrap") { - if (!params.nodes || params.nodes.length === 0) - return fail('action "wrap" requires "nodes" (array of node names).'); - if (!params.wrapper) - return fail('action "wrap" requires "wrapper" (container node definition with name and do).'); - if (!["loop", "condition", "branch", "parallel"].includes(params.wrapper.do)) - return fail( - `Wrapper "do" must be loop, condition, branch, or parallel. Got "${params.wrapper.do}".`, - ); - - // All target nodes must be in the same parent list and contiguous - const firstFound = deepFind(params.nodes[0]); - if (!firstFound) - return fail(`Node "${params.nodes[0]}" not found.`); - - const parentList = firstFound.list; - const indices: number[] = []; - for (const name of params.nodes) { - const idx = parentList.findIndex((n) => n.name === name); - if (idx === -1) - return fail( - `Node "${name}" not found in the same parent list as "${params.nodes[0]}".`, - ); - indices.push(idx); - } - indices.sort((a, b) => a - b); - - // Check contiguous - for (let i = 1; i < indices.length; i++) { - if (indices[i] !== indices[i - 1] + 1) - return fail( - "Nodes to wrap must be contiguous (adjacent in the same list).", - ); - } - - // Extract the nodes - const extracted = parentList.splice(indices[0], indices.length); - - // Build the wrapper node with children - const wrapperNode = { ...params.wrapper } as FlowNode; - switch (wrapperNode.do) { - case "loop": - (wrapperNode as LoopNode).nodes = extracted; - break; - case "parallel": - (wrapperNode as ParallelNode).nodes = extracted; - break; - case "condition": - (wrapperNode as ConditionNode).then = extracted; - if (!(wrapperNode as ConditionNode).else) - (wrapperNode as ConditionNode).else = []; - break; - case "branch": { - const b = wrapperNode as BranchNode; - // Put extracted nodes into the first path, or "true" by default - const firstPath = b.paths ? Object.keys(b.paths)[0] : "true"; - if (!b.paths) b.paths = {}; - b.paths[firstPath] = extracted; - break; - } - } - - // Insert wrapper where the first extracted node was - parentList.splice(indices[0], 0, wrapperNode); - - const validation = validateFlow(flowDef); - if (!validation.ok) { - // rollback: remove wrapper, re-insert extracted nodes - parentList.splice(indices[0], 1); - parentList.splice(indices[0], 0, ...extracted); - return fail( - `Validation failed after wrap (rolled back):\n${validation.errors.map((e) => ` - ${e.message}`).join("\n")}`, - ); - } - - if (filePath) { - await saveSnapshot(); - const { writeFileSync } = await import("fs"); - writeFileSync(filePath, JSON.stringify(flowDef, null, 2) + "\n"); - } - return ok( - `Wrapped ${params.nodes.length} node(s) into "${params.wrapper.name}" (${params.wrapper.do}).${filePath ? ` File written: ${filePath}` : ""}`, - flowDef, - ); - } - - return fail(`Unknown action: "${params.action}"`); - }, - }, - { optional: true }, - ); } export default { diff --git a/tests/core.test.ts b/tests/core.test.ts index 7032ffa..3a925dd 100644 --- a/tests/core.test.ts +++ b/tests/core.test.ts @@ -4,8 +4,8 @@ import * as fs from "fs"; import * as path from "path"; import * as os from "os"; -import { FlowRunner, parseDuration, transpileToCloudflare, validateFlow } from "../src/index.js"; -import type { FlowDefinition, PluginConfig } from "../src/index.js"; +import { FlowRunner, parseDuration, transpileToCloudflare, validateFlow, OpenClawCliInvoker } from "../src/index.js"; +import type { FlowDefinition, PluginConfig, AgentInvoker, AgentTarget, AgentInvokeOptions } from "../src/index.js"; // Use a temp dir for state so tests don't pollute the real store const tmpDir = path.join(os.tmpdir(), `ocf-test-${Date.now()}`); @@ -271,21 +271,47 @@ describe("FlowRunner — agent node field resolution", () => { }); it("builds openclaw agent selector args from target fields", () => { - const runner = new FlowRunner(cfg) as any; + const invoker = new OpenClawCliInvoker({ defaultAgent: "main" }) as any; // all selectors pass through in order; --channel is delivery, not a target assert.deepEqual( - runner.buildAgentArgs({ agent: "ops", sessionKey: "s-key", sessionId: "s-id", channel: "slack" }), + invoker.buildArgs({ agent: "ops", sessionKey: "s-key", sessionId: "s-id", channel: "slack" }), ["--agent", "ops", "--session-key", "s-key", "--session-id", "s-id", "--channel", "slack"], ); // sessionKey only, no agent → key alone (openclaw scopes it) - assert.deepEqual(runner.buildAgentArgs({ sessionKey: "agent:main:slack:c1" }), [ + assert.deepEqual(invoker.buildArgs({ sessionKey: "agent:main:slack:c1" }), [ "--session-key", "agent:main:slack:c1", ]); // no target selector at all → fall back to a routable default agent - assert.deepEqual(runner.buildAgentArgs({}), ["--agent", "main"]); - assert.deepEqual(runner.buildAgentArgs({ channel: "slack" }), ["--channel", "slack", "--agent", "main"]); + assert.deepEqual(invoker.buildArgs({}), ["--agent", "main"]); + assert.deepEqual(invoker.buildArgs({ channel: "slack" }), ["--channel", "slack", "--agent", "main"]); // a bare sessionId is a target → no default-agent fallback - assert.deepEqual(runner.buildAgentArgs({ sessionId: "sess-9" }), ["--session-id", "sess-9"]); + assert.deepEqual(invoker.buildArgs({ sessionId: "sess-9" }), ["--session-id", "sess-9"]); + }); + + it("routes agent nodes through a custom AgentInvoker", async () => { + const calls: Array<{ message: string; target: AgentTarget; opts: AgentInvokeOptions }> = []; + const fakeInvoker: AgentInvoker = { + async invoke(message, target, opts) { + calls.push({ message, target, opts }); + return '{"answer": 42}'; + }, + }; + const flow: FlowDefinition = { + flow: "test-agent-invoker", + nodes: [ + { name: "delegate", do: "agent" as const, task: "compute {{ inputs.q }}", agent: "ops", channel: "slack", output: "result" }, + ], + }; + const runner = new FlowRunner({ ...cfg, agentInvoker: fakeInvoker }); + const result = await runner.run(flow, { q: "meaning" }); + assert.equal(result.ok, true); + // reply is auto-parsed as JSON + assert.deepEqual(result.state.result, { answer: 42 }); + assert.equal(calls.length, 1); + assert.equal(calls[0].message, "compute meaning"); + assert.deepEqual(calls[0].target, { agent: "ops", sessionKey: undefined, sessionId: undefined, channel: "slack" }); + // default timeout applies when the node sets none + assert.equal(calls[0].opts.timeoutMs, 120_000); }); }); diff --git a/tests/ops.test.ts b/tests/ops.test.ts new file mode 100644 index 0000000..7481f34 --- /dev/null +++ b/tests/ops.test.ts @@ -0,0 +1,266 @@ +import { describe, it, 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 * as http from "http"; + +import { + FlowRunner, + FlowOps, + FLOW_OP_SPECS, + startManagementServer, +} from "../src/index.js"; +import type { FlowDefinition, PluginConfig } from "../src/index.js"; + +// Temp workspace so tests don't pollute the real one +const tmpDir = path.join(os.tmpdir(), `ocf-ops-test-${Date.now()}`); +const workspace = path.join(tmpDir, "workspace"); +const cfg: PluginConfig = { + stateDir: path.join(tmpDir, "state"), + memoryDir: path.join(tmpDir, "memory"), +}; + +function makeOps(): FlowOps { + return new FlowOps(workspace, new FlowRunner(cfg)); +} + +function cleanup() { + fs.rmSync(tmpDir, { recursive: true, force: true }); +} + +const SIMPLE_FLOW: FlowDefinition = { + flow: "ops-simple", + nodes: [ + { name: "double", do: "code" as const, run: "input * 2", input: "inputs.x", output: "result" }, + ], +}; + +// ---- FlowOps lifecycle ---------------------------------------------------------- + +describe("FlowOps — create / list / read / publish / run / delete / restore", () => { + after(cleanup); + const ops = makeOps(); + + it("creates a flow file", async () => { + const r = await ops.execute("flow_create", { + file: "ops-simple", + flow: "ops-simple", + description: "doubles x", + nodes: SIMPLE_FLOW.nodes, + }); + assert.match(r.text, /created at/); + assert.ok(fs.existsSync(path.join(workspace, "flows", "ops-simple.json"))); + }); + + it("rejects creating over an existing file", async () => { + const r = await ops.execute("flow_create", { + file: "ops-simple", + flow: "ops-simple", + nodes: SIMPLE_FLOW.nodes, + }); + assert.match(r.text, /already exists/); + }); + + it("rejects invalid flows", async () => { + const r = await ops.execute("flow_create", { + file: "ops-bad", + flow: "ops-bad", + nodes: [{ name: "x", do: "nope" }], + }); + assert.match(r.text, /Validation failed/); + assert.ok(!fs.existsSync(path.join(workspace, "flows", "ops-bad.json"))); + }); + + it("lists flows with metadata", async () => { + const r = await ops.execute("flow_list", {}); + const flows = r.details as Array<{ flow: string; nodes: number }>; + assert.equal(flows.length, 1); + assert.equal(flows[0].flow, "ops-simple"); + assert.equal(flows[0].nodes, 1); + }); + + it("reads a flow with expected inputs", async () => { + const r = await ops.execute("flow_read", { file: "ops-simple" }); + const detail = r.details as { flow: string; _source: string }; + assert.equal(detail.flow, "ops-simple"); + assert.equal(detail._source, "draft"); + }); + + it("edits a flow (add node)", async () => { + const r = await ops.execute("flow_edit", { + file: "ops-simple", + action: "add", + nodeDefinition: { name: "plus-one", do: "code", run: "input + 1", input: "result", output: "final" }, + }); + assert.match(r.text, /added at position 1/); + }); + + it("publishes a version", async () => { + const r = await ops.execute("flow_publish", { file: "ops-simple" }); + assert.match(r.text, /as v1/); + assert.ok( + fs.existsSync(path.join(workspace, ".clawflow", "versions", "ops-simple", "1.json")), + ); + }); + + it("runs the published version by default", async () => { + const r = await ops.execute("flow_run", { file: "ops-simple", input: { x: 4 } }); + const out = r.details as { ok: boolean; _source: string; state: { final: number } }; + assert.equal(out.ok, true); + assert.equal(out._source, "v1"); + assert.equal(out.state.final, 9); + }); + + it("runs the draft when asked", async () => { + const r = await ops.execute("flow_run", { file: "ops-simple", input: { x: 4 }, draft: true }); + const out = r.details as { ok: boolean; _source: string }; + assert.equal(out.ok, true); + assert.equal(out._source, "draft"); + }); + + it("reports status for completed instances", async () => { + const r = await ops.execute("flow_status", {}); + const list = r.details as Array<{ status: string }>; + assert.ok(list.length >= 2); + assert.ok(list.every((i) => i.status === "completed")); + }); + + it("deletes to bin and restores", async () => { + const del = await ops.execute("flow_delete", { file: "ops-simple" }); + assert.match(del.text, /moved to bin/); + assert.ok(!fs.existsSync(path.join(workspace, "flows", "ops-simple.json"))); + + const listBin = await ops.execute("flow_restore_from_bin", {}); + const entries = listBin.details as Array<{ name: string }>; + assert.equal(entries[0].name, "ops-simple"); + + const restore = await ops.execute("flow_restore_from_bin", { name: "ops-simple" }); + assert.match(restore.text, /Restored "ops-simple"/); + assert.ok(fs.existsSync(path.join(workspace, "flows", "ops-simple.json"))); + }); + + it("throws on unknown op names", async () => { + await assert.rejects(() => ops.execute("flow_nope", {}), /Unknown flow op/); + }); +}); + +// ---- Op specs ------------------------------------------------------------------- + +describe("FLOW_OP_SPECS", () => { + it("covers every op the dispatcher accepts, exactly once", async () => { + const names = FLOW_OP_SPECS.map((s) => s.name); + assert.equal(new Set(names).size, names.length); + // every spec dispatches (never "Unknown flow op"; missing-param errors are + // fine — the harness validates params against the schema before calling) + const ops = makeOps(); + for (const name of names) { + try { + await ops.execute(name, {}); + } catch (err) { + assert.doesNotMatch(String(err), /Unknown flow op/); + } + } + }); + + it("carries a JSON-schema parameters object on every spec", () => { + for (const spec of FLOW_OP_SPECS) { + assert.equal(typeof spec.description, "string"); + assert.equal((spec.parameters as { type: string }).type, "object"); + } + }); +}); + +// ---- Management server ---------------------------------------------------------- + +describe("management server", () => { + after(cleanup); + + const TOKEN = "test-token-123"; + let server: http.Server; + let port: number; + + async function request( + method: string, + pathname: string, + opts: { token?: string; body?: unknown } = {}, + ): Promise<{ status: number; body: any }> { + const res = await fetch(`http://127.0.0.1:${port}${pathname}`, { + method, + headers: { + ...(opts.token && { Authorization: `Bearer ${opts.token}` }), + "Content-Type": "application/json", + }, + ...(opts.body !== undefined && { body: JSON.stringify(opts.body) }), + }); + return { status: res.status, body: await res.json() }; + } + + it("starts on an ephemeral port", async () => { + server = startManagementServer({ ops: makeOps(), token: TOKEN, port: 0 }); + await new Promise((resolve) => server.on("listening", resolve)); + const addr = server.address(); + assert.ok(typeof addr === "object" && addr); + port = (addr as { port: number }).port; + }); + + it("rejects requests without a token", async () => { + const r = await request("GET", "/ops"); + assert.equal(r.status, 401); + }); + + it("rejects requests with a wrong token", async () => { + const r = await request("GET", "/ops", { token: "wrong" }); + assert.equal(r.status, 401); + }); + + it("serves op specs on GET /ops", async () => { + const r = await request("GET", "/ops", { token: TOKEN }); + assert.equal(r.status, 200); + assert.equal(r.body.ops.length, FLOW_OP_SPECS.length); + assert.ok(r.body.ops.some((s: { name: string }) => s.name === "flow_run")); + }); + + it("answers health", async () => { + const r = await request("GET", "/ops/health", { token: TOKEN }); + assert.equal(r.status, 200); + assert.equal(r.body.ok, true); + }); + + it("executes an op end-to-end (create → run)", async () => { + const create = await request("POST", "/ops/flow_create", { + token: TOKEN, + body: { file: "manage-flow", flow: "manage-flow", nodes: SIMPLE_FLOW.nodes }, + }); + assert.equal(create.status, 200); + assert.match(create.body.text, /created at/); + + const run = await request("POST", "/ops/flow_run", { + token: TOKEN, + body: { file: "manage-flow", input: { x: 21 } }, + }); + assert.equal(run.status, 200); + assert.equal(run.body.details.ok, true); + assert.equal(run.body.details.state.result, 42); + }); + + it("404s unknown ops", async () => { + const r = await request("POST", "/ops/flow_nope", { token: TOKEN, body: {} }); + assert.equal(r.status, 404); + }); + + it("400s invalid JSON bodies", async () => { + const res = await fetch(`http://127.0.0.1:${port}/ops/flow_list`, { + method: "POST", + headers: { Authorization: `Bearer ${TOKEN}` }, + body: "{not json", + }); + assert.equal(res.status, 400); + }); + + it("stops cleanly", async () => { + await new Promise((resolve, reject) => + server.close((err) => (err ? reject(err) : resolve())), + ); + }); +});