flow_run fails an agent step that targets the session running it, instead of waiting until the step times out (1.7.1)

package.json

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

skills/clawflow/SKILL.md

@@ -56,6 +56,7 @@ A flow is JSON with a `flow` name, an optional `env` block, and a `nodes` array.
 - Set `agent: "clawflow"` on agent nodes to target a configured OpenClaw agent (a plain slug like `main`/`clawflow`, never a session key)
 - Set `sessionKey: "agent:main:slack:channel:agent"` on agent nodes to run inside a specific existing session (e.g. a channel) — maps to `--session-key`. An `agent:`-prefixed key is self-scoping; a bare key (e.g. `incident-42`) is scoped by `agent`. May be combined with `agent`
 - Set `sessionId: "sess-9"` to target one explicit session by id (maps to `--session-id`) — the most specific selector, scoped by `agent`. Use `sessionKey` for channel/self-scoping keys, `sessionId` to resume a known id
+- An agent step that targets a session waits for that session to be free: its turn queues behind any turn already running there. Never `flow_run` such a flow from the session it targets. `flow_run` keeps your turn busy until the flow ends, so the step could never start; the runner fails it at once to say so. Test it by firing its real trigger while that session is idle
 - Set `channel: "slack"` to deliver the agent's reply on a specific channel (maps to `--channel`); omit to use the session's own channel
 - `agent`, `sessionKey`, `sessionId`, and `channel` all support templates, so a prior node can compute the target (e.g. `agent: "{{ route.slug }}"`)
 - Use `do: wait` with `for: approval` before any side effects that need human review — it pauses the flow, provides a token, and shows preview data to the approver

src/core/runner.ts

@@ -125,10 +125,31 @@ export class FlowRunner {
 
   // ---- Start a new run ----------------------------------------------------------
 
+  /** The agent session that started a run and is waiting on it (flow_run).
+   *  An agent step targeting that session could only start once the waiting
+   *  turn ends, so the run would deadlock until the step times out. */
+  private callers = new Map<string, CallerSession>();
+
   async run(
     flow: FlowDefinition,
     input: unknown,
     instanceId?: string,
+    opts?: { caller?: CallerSession },
+  ): Promise<FlowResult> {
+    if (!opts?.caller?.sessionKey && !opts?.caller?.sessionId) return this.runFlow(flow, input, instanceId);
+    const id = instanceId ?? crypto.randomUUID();
+    this.callers.set(id, opts.caller);
+    try {
+      return await this.runFlow(flow, input, id);
+    } finally {
+      this.callers.delete(id);
+    }
+  }
+
+  private async runFlow(
+    flow: FlowDefinition,
+    input: unknown,
+    instanceId?: string,
   ): Promise<FlowResult> {
     // Static validation before execution
     const validation = validateFlow(flow, { registry: this.registry });
@@ -520,7 +541,7 @@ export class FlowRunner {
       case "ai":
         return this.execAi(node as AiNode, state);
       case "agent":
-        return this.execAgent(node as AgentNode, state);
+        return this.execAgent(node as AgentNode, state, instanceId);
       case "branch":
         return this.execBranch(node as BranchNode, state, flow, instanceId);
       case "condition":
@@ -974,6 +995,7 @@ export class FlowRunner {
   private async execAgent(
     node: AgentNode,
     state: FlowState,
+    instanceId: string,
   ): Promise<{ output: unknown }> {
     const task = node.task; // template already interpolated by resolveNodeFields
     const input = node.input ? this.getPath(state, node.input) : undefined;
@@ -986,18 +1008,23 @@ export class FlowRunner {
       ? parseDuration(node.timeout)
       : undefined;
 
-    const cliResult = await this.tryOpenClawAgent(
-      fullPrompt,
-      {
-        // new names win; agentId/session are accepted as deprecated aliases
-        agent: node.agent ?? node.agentId,
-        sessionKey: node.sessionKey ?? node.session,
-        sessionId: node.sessionId,
-        channel: node.channel,
-      },
-      state,
-      timeoutMs,
-    );
+    const target = {
+      // new names win; agentId/session are accepted as deprecated aliases
+      agent: node.agent ?? node.agentId,
+      sessionKey: node.sessionKey ?? node.session,
+      sessionId: node.sessionId,
+      channel: node.channel,
+    };
+    const caller = this.callers.get(instanceId);
+    if (caller && targetsCallerSession(target, caller, this.cfg.defaultAgent)) {
+      throw new Error(
+        `This step runs in the same session as the agent that started this flow with flow_run. ` +
+          `That session is busy until flow_run returns, so the step could never start and would wait until it timed out. ` +
+          `Test the flow by triggering it for real while that session is idle, or point the step at another session.`,
+      );
+    }
+
+    const cliResult = await this.tryOpenClawAgent(fullPrompt, target, state, timeoutMs);
     return { output: this.autoParseJson(cliResult) };
   }
 
@@ -2006,3 +2033,27 @@ export class FlowRunner {
     return this.store.listApprovals();
   }
 }
+
+/** The OpenClaw session a run was started from (the flow_run tool's caller). */
+export interface CallerSession {
+  sessionKey?: string;
+  sessionId?: string;
+}
+
+/** Whether an agent step's resolved target is the caller's own session. Keys
+ *  are compared the way OpenClaw canonicalises them (session-store-key.ts,
+ *  canonicalizeSessionKeyForAgent): lower-cased, then a key without the
+ *  `agent:` prefix is scoped by the step's agent. A step that names only an
+ *  agent runs in that agent's default session, which is not resolved here, so
+ *  it is not caught. */
+export function targetsCallerSession(
+  target: { agent?: string; sessionKey?: string; sessionId?: string },
+  caller: CallerSession,
+  defaultAgent?: string,
+): boolean {
+  if (target.sessionId && caller.sessionId && target.sessionId === caller.sessionId) return true;
+  if (!target.sessionKey || !caller.sessionKey) return false;
+  const raw = target.sessionKey.toLowerCase();
+  const key = raw.startsWith("agent:") ? raw : `agent:${(target.agent ?? defaultAgent ?? "main").toLowerCase()}:${raw}`;
+  return key === caller.sessionKey.toLowerCase();
+}

src/index.ts

@@ -5,7 +5,8 @@
 // standalone consumers (e.g. `@clawnify/clawflow` as a Cloudflare runtime).
 export { default } from "./plugin/index.js";
 
-export { FlowRunner, sendEvent } from "./core/runner.js";
+export { FlowRunner, sendEvent, targetsCallerSession } from "./core/runner.js";
+export type { CallerSession } from "./core/runner.js";
 export { StateStore } from "./core/store.js";
 export { transpileToCloudflare } from "./core/transpile.js";
 export { validateFlow } from "./core/validate.js";

src/plugin/index.ts

@@ -142,8 +142,11 @@ type RequireApprovalDecision = {
   timeoutBehavior?: "allow" | "deny";
 };
 
+/** The per-call context OpenClaw hands a tool factory (OpenClawPluginToolContext). */
+type ToolContext = { sessionKey?: string; sessionId?: string };
+
 interface PluginApi {
-  registerTool: (def: object, opts?: { optional?: boolean }) => void;
+  registerTool: (def: object | ((ctx?: ToolContext) => object), opts?: { name?: string; optional?: boolean }) => void;
   registerHook?: (
     events: string | string[],
     handler: (event: BeforeToolCallEvent) =>
@@ -648,7 +651,10 @@ directory. If the flow file already exists, the restore is rejected.`,
   // ---- flow_run -----------------------------------------------------------------
 
   api.registerTool(
-    {
+    // A factory, so the tool knows which session called it: flow_run blocks
+    // that session until the flow ends, and the runner refuses an agent step
+    // that targets it instead of letting the step wait until it times out.
+    (ctx?: ToolContext) => ({
       name: "flow_run",
       catalogMode: "direct-only",
       description: `Run an agentic workflow in the clawflow format.
@@ -799,7 +805,9 @@ Set version to run a specific published version.`,
         }
 
         try {
-          const result = await runner.run(flowDef, params.input ?? {});
+          const result = await runner.run(flowDef, params.input ?? {}, undefined, {
+            caller: { sessionKey: ctx?.sessionKey, sessionId: ctx?.sessionId },
+          });
           const out = { ...result, _source: source };
           // Whole when it fits, else a summary the agent reads on from with
           // flow_status { instanceId, path } (see core/runs.ts).
@@ -820,8 +828,8 @@ Set version to run a specific published version.`,
           };
         }
       },
-    },
-    { optional: true },
+    }),
+    { name: "flow_run", optional: true },
   );
 
   // ---- flow_resume --------------------------------------------------------------

tests/core.test.ts

@@ -4,7 +4,7 @@ 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 { FlowRunner, parseDuration, targetsCallerSession, transpileToCloudflare, validateFlow } from "../src/index.js";
 import type { FlowDefinition, PluginConfig } from "../src/index.js";
 
 // Use a temp dir for state so tests don't pollute the real store
@@ -289,6 +289,65 @@ describe("FlowRunner — agent node field resolution", () => {
   });
 });
 
+describe("FlowRunner: an agent step targeting the flow_run caller's session", () => {
+  after(cleanup);
+
+  // A flow whose agent step replies in a chat, run with flow_run from that same
+  // chat: flow_run keeps the chat's turn busy until the flow ends, and the
+  // step's turn would queue behind it, so both would wait until the step's
+  // timeout.
+  const chat = "agent:main:user:u1:chat:c1";
+  const flow: FlowDefinition = {
+    flow: "ack",
+    nodes: [{ name: "ack_in_chat", do: "agent" as const, sessionKey: chat, task: "Reply OK" }],
+  };
+
+  function stubbedRunner() {
+    const runner = new FlowRunner(cfg) as any;
+    const calls: unknown[] = [];
+    runner.tryOpenClawAgent = async (_message: string, target: unknown) => {
+      calls.push(target);
+      return "OK";
+    };
+    return { runner, calls };
+  }
+
+  it("fails the step at once instead of starting a turn it would wait on", async () => {
+    const { runner, calls } = stubbedRunner();
+    const result = await runner.run(flow, {}, undefined, { caller: { sessionKey: chat } });
+    assert.equal(result.ok, false);
+    assert.match(result.error ?? "", /same session as the agent that started this flow/);
+    assert.equal(calls.length, 0, "the openclaw agent CLI must not be called");
+  });
+
+  it("runs the step when the caller is another session, or there is no caller", async () => {
+    const other = stubbedRunner();
+    const fromOther = await other.runner.run(flow, {}, undefined, { caller: { sessionKey: "agent:main:main" } });
+    assert.equal(fromOther.ok, true);
+    assert.equal(other.calls.length, 1);
+
+    const trigger = stubbedRunner();
+    const fromTrigger = await trigger.runner.run(flow, {});
+    assert.equal(fromTrigger.ok, true);
+    assert.equal(trigger.calls.length, 1);
+  });
+
+  it("matches keys the way openclaw scopes them", () => {
+    const caller = { sessionKey: chat, sessionId: "sess-1" };
+    assert.equal(targetsCallerSession({ sessionKey: chat }, caller), true);
+    assert.equal(targetsCallerSession({ sessionKey: chat.toUpperCase() }, caller), true);
+    // a bare key is scoped by the step's agent, else the default agent
+    assert.equal(targetsCallerSession({ sessionKey: "user:u1:chat:c1" }, caller), true);
+    assert.equal(targetsCallerSession({ agent: "ops", sessionKey: "user:u1:chat:c1" }, caller), false);
+    assert.equal(targetsCallerSession({ sessionKey: "user:u1:chat:c1" }, caller, "ops"), false);
+    assert.equal(targetsCallerSession({ sessionId: "sess-1" }, caller), true);
+    assert.equal(targetsCallerSession({ sessionKey: "agent:main:user:u1:chat:c2" }, caller), false);
+    // a step that names only an agent runs in a session not resolved here
+    assert.equal(targetsCallerSession({ agent: "main" }, caller), false);
+    assert.equal(targetsCallerSession({ sessionKey: chat }, {}), false);
+  });
+});
+
 describe("FlowRunner — exec node", () => {
   after(cleanup);