Harness-neutral core: FlowOps + op specs as data + management API + AgentInvoker

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)

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@clawnify/clawflow",
-  "version": "1.4.1",
+  "version": "1.5.0",
   "description": "The n8n for agents. A declarative, AI-native workflow format that agents can read, write, and run.",
   "type": "module",
   "main": "./dist/index.js",

src/core/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<string, string>;
+}
+
+export interface AgentInvoker {
+  /** Run one agent turn and return the agent's reply as text. */
+  invoke(
+    message: string,
+    target: AgentTarget,
+    opts: AgentInvokeOptions,
+  ): Promise<string>;
+}
+
+// ---- 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 <id>         scopes to a configured agent
+  //   --session-key <key>  an exact session key (agent:<id>:<key>, or scoped to --agent)
+  //   --session-id <id>    the most specific: one explicit session (scoped by --agent)
+  //   --channel <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<string> {
+    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}`);
+    }
+  }
+}

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<string> {
+  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;
+}