Record the step a run is executing on its record, cleared when the step ends (1.8.0)

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@clawnify/clawflow",
-  "version": "1.7.1",
+  "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",

src/core/runner.ts

@@ -8,6 +8,7 @@ import {
   FlowState,
   FlowResult,
   TraceEntry,
+  RunningStep,
   PluginConfig,
   MODEL_MAP,
   OPENROUTER_MODEL_MAP,
@@ -373,6 +374,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) ----------------------
@@ -388,6 +391,7 @@ export class FlowRunner {
             status: "paused",
             state,
             trace,
+            running: undefined,
             pausedAtIndex: i,
             resumeToken: result.approve.token,
             waitingFor: {
@@ -427,7 +431,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);
@@ -440,7 +444,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",
@@ -477,9 +481,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");
 
@@ -512,6 +514,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")

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<string, unknown>; // 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"];

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"

src/index.ts

@@ -35,6 +35,7 @@ export type {
   FlowStatus,
   InputSpec,
   TraceEntry,
+  RunningStep,
   NodeStatus,
   RetryPolicy,
   PluginConfig,

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: an agent step targeting the flow_run caller's session", () => {
   after(cleanup);