From dc6c1e92fc1430a27d59a43cb882f297dd769b7f Mon Sep 17 00:00:00 2001 From: Riccardo Pallaoro <95774148+pallaoro@users.noreply.github.com> Date: Wed, 4 Nov 2026 17:57:07 +0100 Subject: [PATCH] Record the step a run is executing on its record, cleared when the step ends (1.8.0) The trace gains an entry only when a step ends, so a running flow showed nothing about what it was doing. The run record now carries running: the step's name and type, when it started, its per-attempt timeout, and for an agent step the resolved session it runs in. Each exit from a step clears it in the same write that records the step's end. --- package-lock.json | 4 ++-- package.json | 2 +- src/core/runner.ts | 32 +++++++++++++++++++++++++++----- src/core/store.ts | 3 ++- src/core/types.ts | 15 +++++++++++++++ src/index.ts | 1 + tests/core.test.ts | 38 ++++++++++++++++++++++++++++++++++++++ 7 files changed, 86 insertions(+), 9 deletions(-) diff --git a/package-lock.json b/package-lock.json index a57b1c4..dade0ef 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@clawnify/clawflow", - "version": "1.7.0", + "version": "1.8.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@clawnify/clawflow", - "version": "1.7.0", + "version": "1.8.0", "license": "MIT", "dependencies": { "croner": "^10.0.1" diff --git a/package.json b/package.json index d790006..d8d20d2 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@clawnify/clawflow", - "version": "1.7.0", + "version": "1.8.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/runner.ts b/src/core/runner.ts index f883f7c..ded8e74 100644 --- a/src/core/runner.ts +++ b/src/core/runner.ts @@ -8,6 +8,7 @@ import { FlowState, FlowResult, TraceEntry, + RunningStep, PluginConfig, MODEL_MAP, OPENROUTER_MODEL_MAP, @@ -352,6 +353,8 @@ export class FlowRunner { } try { + // Say what this run is doing while the step runs; each exit below clears it. + this.store.update(instanceId, { running: this.runningStep(node, state) }); const result = await this.runWithRetry(node, state, flow, instanceId); // ---- Paused for approval (do: wait, for: approval) ---------------------- @@ -367,6 +370,7 @@ export class FlowRunner { status: "paused", state, trace, + running: undefined, pausedAtIndex: i, resumeToken: result.approve.token, waitingFor: { @@ -406,7 +410,7 @@ export class FlowRunner { durationMs: Date.now() - t0, }; trace.push(entry); - this.store.update(instanceId, { state, trace }); + this.store.update(instanceId, { state, trace, running: undefined }); i++; } catch (err) { const message = err instanceof Error ? err.message : String(err); @@ -419,7 +423,7 @@ export class FlowRunner { }; trace.push(entry); const error = `Node "${node.name}": ${message}`; - this.store.update(instanceId, { status: "failed", state, trace, error }); + this.store.update(instanceId, { status: "failed", state, trace, error, running: undefined }); return { ok: false, status: "failed", @@ -456,9 +460,7 @@ export class FlowRunner { attempts?: number; }> { const policy: RetryPolicy = node.retry ?? { limit: 1, delay: 0 }; - const nodeTimeoutMs = node.timeout - ? parseDuration(node.timeout) - : (this.cfg.maxNodeDurationMs ?? 30_000); + const nodeTimeoutMs = this.nodeTimeoutMs(node); let lastError: Error = new Error("Unknown error"); @@ -491,6 +493,26 @@ export class FlowRunner { throw lastError; } + /** How long one attempt of a step may run before it fails. */ + private nodeTimeoutMs(node: FlowNode): number { + return node.timeout ? parseDuration(node.timeout) : (this.cfg.maxNodeDurationMs ?? 30_000); + } + + private runningStep(node: FlowNode, state: FlowState): RunningStep { + const step: RunningStep = { + node: node.name, + do: node.do, + startedAt: new Date().toISOString(), + timeoutMs: this.nodeTimeoutMs(node), + }; + if (node.do === "agent") { + // The same interpolation execNode applies, so this is the session the step runs in. + const n = this.resolveNodeFields(node, state) as AgentNode; + step.target = { agent: n.agent ?? n.agentId, sessionKey: n.sessionKey ?? n.session, sessionId: n.sessionId }; + } + return step; + } + private calcDelay(policy: RetryPolicy, attempt: number): number { const base = parseDuration(policy.delay); if (policy.backoff === "exponential") diff --git a/src/core/store.ts b/src/core/store.ts index 002f9e6..ff50804 100644 --- a/src/core/store.ts +++ b/src/core/store.ts @@ -1,6 +1,6 @@ import * as fs from "fs"; import * as path from "path"; -import type { FlowState, FlowResult, TraceEntry, PendingApproval } from "./types.js"; +import type { FlowState, FlowResult, TraceEntry, PendingApproval, RunningStep } from "./types.js"; // ---- Durable State Store -------------------------------------------------------- // Persists flow instance state to disk so flows survive gateway restarts. @@ -22,6 +22,7 @@ export interface InstanceRecord { state: FlowState; completedNodes: Record; // nodeName -> output (memoized) trace: TraceEntry[]; // persisted trace survives resume + running?: RunningStep; // the step executing now; cleared when it ends pausedAtIndex?: number; // node index where flow paused (for resume) resumeToken?: string; waitingFor?: FlowResult["waitingFor"]; diff --git a/src/core/types.ts b/src/core/types.ts index 561fc86..79a1afb 100644 --- a/src/core/types.ts +++ b/src/core/types.ts @@ -413,6 +413,21 @@ export interface TraceEntry { durationMs: number; } +/** The top-level step a run is executing now. The trace only gains an entry + * when a step ends, so without this a running flow shows nothing about what + * it is doing or what it is waiting on. */ +export interface RunningStep { + node: string; + do: string; + startedAt: string; + /** How long one attempt may run before it fails, so a reader can say how + * long the step may still wait. */ + timeoutMs: number; + /** Agent steps: the resolved session the step runs in. Its turn waits for + * any turn already running in that session. */ + target?: { agent?: string; sessionKey?: string; sessionId?: string }; +} + export type FlowStatus = | "running" | "completed" diff --git a/src/index.ts b/src/index.ts index ff0fa7b..156883a 100644 --- a/src/index.ts +++ b/src/index.ts @@ -34,6 +34,7 @@ export type { FlowStatus, InputSpec, TraceEntry, + RunningStep, NodeStatus, RetryPolicy, PluginConfig, diff --git a/tests/core.test.ts b/tests/core.test.ts index 7032ffa..a10cb8e 100644 --- a/tests/core.test.ts +++ b/tests/core.test.ts @@ -289,6 +289,44 @@ describe("FlowRunner — agent node field resolution", () => { }); }); +describe("FlowRunner: the step a run is executing, on its record", () => { + after(cleanup); + + const flow: FlowDefinition = { + flow: "step", + nodes: [{ name: "ask", do: "agent" as const, sessionKey: "agent:main:{{ inputs.chat }}", task: "Reply OK", timeout: "90s" }], + }; + + it("records the step while it runs, with an agent step's resolved session, and clears it after", async () => { + const runner = new FlowRunner(cfg) as any; + let seen: any; + runner.tryOpenClawAgent = async () => { + seen = runner.getStore().get("run-step-ok")?.running; + return "OK"; + }; + const result = await runner.run(flow, { chat: "c1" }, "run-step-ok"); + assert.equal(result.ok, true); + assert.equal(seen?.node, "ask"); + assert.equal(seen?.do, "agent"); + assert.equal(seen?.timeoutMs, 90_000); + assert.deepEqual(seen?.target, { sessionKey: "agent:main:c1" }); + assert.ok(!Number.isNaN(Date.parse(seen?.startedAt)), "startedAt is a timestamp"); + assert.equal(runner.getStore().get("run-step-ok").running, undefined); + }); + + it("clears it when the step fails", async () => { + const runner = new FlowRunner(cfg) as any; + runner.tryOpenClawAgent = async () => { + throw new Error("boom"); + }; + const result = await runner.run(flow, { chat: "c1" }, "run-step-fail"); + assert.equal(result.ok, false); + const record = runner.getStore().get("run-step-fail"); + assert.equal(record.status, "failed"); + assert.equal(record.running, undefined); + }); +}); + describe("FlowRunner — exec node", () => { after(cleanup);