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