From a6cbc6ae8322773b86ba8321e0eb766c9e41202c Mon Sep 17 00:00:00 2001 From: Riccardo Pallaoro <95774148+pallaoro@users.noreply.github.com> Date: Wed, 4 Nov 2026 17:38:46 +0100 Subject: [PATCH] flow_run fails an agent step that targets the session running it, instead of waiting until the step times out (1.7.1) flow_run keeps the calling session's turn busy until the flow ends. An agent step that targets that same session queues its turn behind it, so the step could never start, and both waited until the step's timeout. - flow_run is registered as a tool factory, so it gets the calling session (sessionKey, sessionId) from the tool context and hands it to the runner. - The runner fails such a step at once, with a message saying why and how to test the flow instead. Keys are compared the way OpenClaw canonicalises them: lower-cased, and a key without the agent: prefix is scoped by the step's agent. A step that names only an agent is not caught. - Runs started any other way (triggers, the flow server) are unaffected. - SKILL.md: do not flow_run such a flow from the session it targets. - Tests: the refusal, the unaffected paths, and key matching. --- package-lock.json | 4 +-- package.json | 2 +- skills/clawflow/SKILL.md | 1 + src/core/runner.ts | 77 +++++++++++++++++++++++++++++++++------- src/index.ts | 3 +- src/plugin/index.ts | 18 +++++++--- tests/core.test.ts | 61 ++++++++++++++++++++++++++++++- 7 files changed, 143 insertions(+), 23 deletions(-) diff --git a/package-lock.json b/package-lock.json index a57b1c4..ac28dcc 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@clawnify/clawflow", - "version": "1.7.0", + "version": "1.7.1", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@clawnify/clawflow", - "version": "1.7.0", + "version": "1.7.1", "license": "MIT", "dependencies": { "croner": "^10.0.1" diff --git a/package.json b/package.json index d790006..d00ceef 100644 --- a/package.json +++ b/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", diff --git a/skills/clawflow/SKILL.md b/skills/clawflow/SKILL.md index 63a3b39..3a24f7a 100644 --- a/skills/clawflow/SKILL.md +++ b/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 diff --git a/src/core/runner.ts b/src/core/runner.ts index f883f7c..e0526e8 100644 --- a/src/core/runner.ts +++ b/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(); + async run( flow: FlowDefinition, input: unknown, instanceId?: string, + opts?: { caller?: CallerSession }, + ): Promise { + 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 { // 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(); +} diff --git a/src/index.ts b/src/index.ts index ff0fa7b..2d02c87 100644 --- a/src/index.ts +++ b/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"; diff --git a/src/plugin/index.ts b/src/plugin/index.ts index de5944e..2760d5b 100644 --- a/src/plugin/index.ts +++ b/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 -------------------------------------------------------------- diff --git a/tests/core.test.ts b/tests/core.test.ts index 7032ffa..52da4d2 100644 --- a/tests/core.test.ts +++ b/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);