Skip to content

Commit 30ea58e

Browse files
jiggai-automation-botjiggai-automation-bot
committed
feat(workflows): add v1 if + delay nodes (decomposed)
Add first-class if and delay workflow node kinds while maintaining the decomposed module structure from PR #183. if node: - Evaluates action.{lhs, op, rhs} using dot-path references to prior node outputs - Supported operators: truthy, ==, !=, >, >=, <, <=, contains - Edges can use on: 'true' | 'false' for conditional branching - Writes boolean output to node-outputs/ delay node: - Pauses run for action.delaySeconds or action.delayMs (max 7 days) - Sets run status to 'paused' with resumeAt timestamp - Runner/tick automatically resumes when resumeAt passes - No new daemon needed — uses existing runner loop cadence Changes by module: - workflow-if.ts: NEW — evalIfCondition + lastIfValueFromRun - workflow-types.ts: add 'if' | 'delay' to NodeKind, 'true' | 'false' to EdgeOn, resumeAt to RunLog - workflow-utils.ts: pause semantics + true/false edge handling in pickNextRunnableNodeIndex - workflow-node-executor.ts: if + delay node execution in inline runner path - workflow-worker.ts: if + delay node execution in pull-based worker path + posting patch - workflow-tick.ts: pick up paused runs for resume - workflow-runner.ts: re-export workflow-if (thin orchestrator preserved) 264/264 tests pass, 0 lint errors.
1 parent 51f3d88 commit 30ea58e

8 files changed

Lines changed: 447 additions & 10 deletions

File tree

‎src/lib/workflows/__tests__/workflow-if.test.ts‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
import { describe, expect, test } from 'vitest';
2+
import fs from 'node:fs/promises';
3+
import path from 'node:path';
4+
import os from 'node:os';
5+
import { evalIfCondition } from '../workflow-if';
6+
7+
async function tmpRunDir() {
8+
const dir = await fs.mkdtemp(path.join(os.tmpdir(), 'clawrecipes-run-'));
9+
await fs.mkdir(path.join(dir, 'node-outputs'), { recursive: true });
10+
return dir;
11+
}
12+
13+
describe('workflow if node', () => {
14+
test('truthy on node output path', async () => {
15+
const runDir = await tmpRunDir();
16+
const outFile = path.join(runDir, 'node-outputs', '000-prev.json');
17+
await fs.writeFile(outFile, JSON.stringify({ text: JSON.stringify({ ok: true, count: 2 }) }, null, 2));
18+
19+
const res = await evalIfCondition({
20+
runDir,
21+
condition: { lhs: 'nodes.prev.output.ok', op: 'truthy' },
22+
});
23+
24+
expect(res.ok).toBe(true);
25+
expect(res.value).toBe(true);
26+
});
27+
28+
test('numeric comparator', async () => {
29+
const runDir = await tmpRunDir();
30+
const outFile = path.join(runDir, 'node-outputs', '000-prev.json');
31+
await fs.writeFile(outFile, JSON.stringify({ text: JSON.stringify({ count: 2 }) }, null, 2));
32+
33+
const res = await evalIfCondition({
34+
runDir,
35+
condition: { lhs: 'nodes.prev.output.count', op: '>=', rhs: 2 },
36+
});
37+
38+
expect(res.value).toBe(true);
39+
});
40+
41+
test('contains on string', async () => {
42+
const runDir = await tmpRunDir();
43+
const outFile = path.join(runDir, 'node-outputs', '000-prev.json');
44+
await fs.writeFile(outFile, JSON.stringify({ text: JSON.stringify({ msg: 'hello world' }) }, null, 2));
45+
46+
const res = await evalIfCondition({
47+
runDir,
48+
condition: { lhs: 'nodes.prev.output.msg', op: 'contains', rhs: 'world' },
49+
});
50+
51+
expect(res.value).toBe(true);
52+
});
53+
});

‎src/lib/workflows/workflow-if.ts‎

Lines changed: 161 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,161 @@
1+
import fs from 'node:fs/promises';
2+
import path from 'node:path';
3+
4+
function isRecord(v: unknown): v is Record<string, unknown> {
5+
return !!v && typeof v === 'object' && !Array.isArray(v);
6+
}
7+
8+
function asRecord(v: unknown): Record<string, unknown> {
9+
return isRecord(v) ? v : {};
10+
}
11+
12+
function asString(v: unknown, fallback = ''): string {
13+
return typeof v === 'string' ? v : v == null ? fallback : String(v);
14+
}
15+
16+
function parseDotPath(p: string): Array<string> {
17+
return p
18+
.split('.')
19+
.map((s) => s.trim())
20+
.filter(Boolean);
21+
}
22+
23+
function getByPath(obj: unknown, p: string): unknown {
24+
const parts = parseDotPath(p);
25+
let cur: unknown = obj;
26+
for (const part of parts) {
27+
if (Array.isArray(cur)) {
28+
const idx = Number(part);
29+
if (!Number.isFinite(idx)) return undefined;
30+
cur = cur[idx];
31+
continue;
32+
}
33+
if (!isRecord(cur)) return undefined;
34+
cur = (cur as Record<string, unknown>)[part];
35+
}
36+
return cur;
37+
}
38+
39+
async function findNodeOutputFile(opts: { runDir: string; nodeId: string }): Promise<string | null> {
40+
const nodeOutputsDir = path.join(opts.runDir, 'node-outputs');
41+
try {
42+
const files = await fs.readdir(nodeOutputsDir);
43+
// Pick the latest by lexical sort (prefix is numeric padded; good enough).
44+
const matches = files.filter((f) => f.endsWith(`-${opts.nodeId}.json`)).sort();
45+
const pick = matches[matches.length - 1];
46+
return pick ? path.join(nodeOutputsDir, pick) : null;
47+
} catch {
48+
return null;
49+
}
50+
}
51+
52+
export async function loadNodeOutputPayload(opts: { runDir: string; nodeId: string }): Promise<unknown> {
53+
const file = await findNodeOutputFile(opts);
54+
if (!file) return undefined;
55+
const raw = await fs.readFile(file, 'utf8');
56+
const parsed = JSON.parse(raw) as unknown;
57+
58+
// Common shape for llm/tool nodes: { text: "{...json...}" }
59+
const rec = asRecord(parsed);
60+
const text = asString(rec['text']).trim();
61+
if (text) {
62+
try {
63+
return JSON.parse(text);
64+
} catch {
65+
// fall through
66+
}
67+
}
68+
69+
return parsed;
70+
}
71+
72+
export type IfComparator =
73+
| 'truthy'
74+
| '=='
75+
| '!='
76+
| '>'
77+
| '>='
78+
| '<'
79+
| '<='
80+
| 'contains';
81+
82+
export type IfCondition = {
83+
lhs: string;
84+
op: IfComparator;
85+
rhs?: unknown;
86+
};
87+
88+
export async function evalIfCondition(opts: { runDir: string; condition: IfCondition }): Promise<{ ok: true; value: boolean; detail: Record<string, unknown> }> {
89+
const lhsRaw = asString(opts.condition.lhs).trim();
90+
const op = asString(opts.condition.op).trim() as IfComparator;
91+
const rhs = opts.condition.rhs;
92+
93+
// Supported v1 reference format:
94+
// - nodes.<nodeId>.output.<path>
95+
// Anything else is treated as a literal string (for now).
96+
let lhsValue: unknown = undefined;
97+
let source: Record<string, unknown> = { kind: 'literal', lhs: lhsRaw };
98+
99+
const m = lhsRaw.match(/^nodes\.([^.]+)\.output\.(.+)$/);
100+
if (m) {
101+
const nodeId = m[1] ?? '';
102+
const outPath = m[2] ?? '';
103+
const payload = await loadNodeOutputPayload({ runDir: opts.runDir, nodeId });
104+
lhsValue = getByPath(payload, outPath);
105+
source = { kind: 'nodeOutput', nodeId, path: outPath };
106+
} else {
107+
lhsValue = lhsRaw;
108+
}
109+
110+
let value = false;
111+
112+
if (op === 'truthy') {
113+
value = !!lhsValue;
114+
} else if (op === '==') {
115+
value = lhsValue === rhs;
116+
} else if (op === '!=') {
117+
value = lhsValue !== rhs;
118+
} else if (op === '>' || op === '>=' || op === '<' || op === '<=') {
119+
const a = typeof lhsValue === 'number' ? lhsValue : Number(lhsValue);
120+
const b = typeof rhs === 'number' ? rhs : Number(rhs);
121+
if (Number.isFinite(a) && Number.isFinite(b)) {
122+
if (op === '>') value = a > b;
123+
if (op === '>=') value = a >= b;
124+
if (op === '<') value = a < b;
125+
if (op === '<=') value = a <= b;
126+
} else {
127+
value = false;
128+
}
129+
} else if (op === 'contains') {
130+
if (typeof lhsValue === 'string') value = typeof rhs === 'string' && lhsValue.includes(rhs);
131+
else if (Array.isArray(lhsValue)) value = lhsValue.some((x) => x === rhs);
132+
else value = false;
133+
} else {
134+
// Unknown op => false.
135+
value = false;
136+
}
137+
138+
return {
139+
ok: true,
140+
value,
141+
detail: {
142+
source,
143+
op,
144+
rhs,
145+
lhsValue,
146+
},
147+
};
148+
}
149+
150+
export function lastIfValueFromRun(run: { nodeResults?: Array<Record<string, unknown>> }, nodeId: string): boolean | null {
151+
const results = Array.isArray(run.nodeResults) ? run.nodeResults : [];
152+
for (let i = results.length - 1; i >= 0; i--) {
153+
const r = asRecord(results[i]);
154+
if (asString(r['nodeId']).trim() !== nodeId) continue;
155+
if (asString(r['kind']).trim() !== 'if') continue;
156+
const v = r['value'];
157+
if (typeof v === 'boolean') return v;
158+
break;
159+
}
160+
return null;
161+
}

‎src/lib/workflows/workflow-node-executor.ts‎

Lines changed: 71 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import { outboundPublish, type OutboundApproval, type OutboundMedia, type Outbou
99
import { sanitizeOutboundPostText } from './outbound-sanitize';
1010
import { loadPriorLlmInput, loadProposedPostTextFromPriorNode } from './workflow-node-output-readers';
1111
import { readTextFile } from './workflow-runner-io';
12+
import { evalIfCondition, lastIfValueFromRun } from './workflow-if';
1213
import {
1314
asRecord, asString,
1415
ensureDir, fileExists,
@@ -75,6 +76,14 @@ export async function executeWorkflowNodes(opts: {
7576
const from = nodeStates[fromId]?.status;
7677
const on = String(e.on ?? 'success');
7778
if (!from) return false;
79+
80+
if (on === 'true' || on === 'false') {
81+
if (from !== 'success') return false;
82+
const v = lastIfValueFromRun(curRun, fromId);
83+
if (v === null) return false;
84+
return on === 'true' ? v === true : v === false;
85+
}
86+
7887
if (on === 'always') return from === 'success' || from === 'error';
7988
if (on === 'error') return from === 'error';
8089
return from === 'success';
@@ -161,6 +170,68 @@ export async function executeWorkflowNodes(opts: {
161170
continue;
162171
}
163172

173+
if (kind === 'if') {
174+
const runDir = path.dirname(runLogPath);
175+
const action = asRecord(node.action);
176+
const lhs = asString(action['lhs']).trim();
177+
const op = asString(action['op']).trim();
178+
const rhs = action['rhs'];
179+
if (!lhs) throw new Error(`Node ${nodeLabel(node)} missing action.lhs`);
180+
if (!op) throw new Error(`Node ${nodeLabel(node)} missing action.op`);
181+
182+
const evalRes = await evalIfCondition({ runDir, condition: { lhs, op: op as 'truthy', rhs } });
183+
184+
const defaultNodeOutputRel = path.join('node-outputs', `${String(i).padStart(3, '0')}-${node.id}.json`);
185+
const nodeOutputRel = String(node?.output?.path ?? '').trim() || defaultNodeOutputRel;
186+
const nodeOutputAbs = path.resolve(path.dirname(runLogPath), nodeOutputRel);
187+
await ensureDir(path.dirname(nodeOutputAbs));
188+
await fs.writeFile(nodeOutputAbs, JSON.stringify({
189+
runId, teamId, nodeId: node.id, kind: node.kind,
190+
completedAt: new Date().toISOString(), value: evalRes.value, detail: evalRes.detail,
191+
}, null, 2) + '\n', 'utf8');
192+
193+
const completedTs = new Date().toISOString();
194+
await appendRunLog(runLogPath, (cur) => ({
195+
...cur,
196+
nextNodeIndex: i + 1,
197+
nodeStates: { ...(cur.nodeStates ?? {}), [node.id]: { status: 'success', ts: completedTs } },
198+
events: [...cur.events, { ts: completedTs, type: 'node.completed', nodeId: node.id, kind, value: evalRes.value, nodeOutputPath: path.relative(teamDir, nodeOutputAbs) }],
199+
nodeResults: [...(cur.nodeResults ?? []), { nodeId: node.id, kind, value: evalRes.value, nodeOutputPath: path.relative(teamDir, nodeOutputAbs) }],
200+
}));
201+
nodeStates[String(node.id)] = { status: 'success', ts: completedTs };
202+
continue;
203+
}
204+
205+
if (kind === 'delay') {
206+
const action = asRecord(node.action);
207+
const secondsRaw = action['seconds'] ?? action['delaySeconds'] ?? action['durationSeconds'];
208+
const msRaw = action['ms'] ?? action['delayMs'] ?? action['durationMs'];
209+
const sec = typeof secondsRaw === 'number' ? secondsRaw : Number(secondsRaw);
210+
const ms = typeof msRaw === 'number' ? msRaw : Number(msRaw);
211+
const delayMs = Number.isFinite(ms) && ms > 0 ? ms : Number.isFinite(sec) && sec > 0 ? sec * 1000 : 0;
212+
if (!delayMs) throw new Error(`Node ${nodeLabel(node)} missing delay duration (action.delaySeconds or action.delayMs)`);
213+
214+
const maxDelayMs = 7 * 24 * 60 * 60 * 1000;
215+
const effectiveDelayMs = Math.min(delayMs, maxDelayMs);
216+
const resumeAt = new Date(Date.now() + effectiveDelayMs).toISOString();
217+
218+
const completedTs = new Date().toISOString();
219+
await appendRunLog(runLogPath, (cur) => ({
220+
...cur,
221+
status: 'paused',
222+
resumeAt,
223+
nextNodeIndex: i + 1,
224+
nodeStates: { ...(cur.nodeStates ?? {}), [node.id]: { status: 'success', ts: completedTs } },
225+
events: [
226+
...cur.events,
227+
{ ts: completedTs, type: 'node.completed', nodeId: node.id, kind, delayMs: effectiveDelayMs, resumeAt },
228+
{ ts: completedTs, type: 'run.paused', nodeId: node.id, resumeAt },
229+
],
230+
nodeResults: [...(cur.nodeResults ?? []), { nodeId: node.id, kind, delayMs: effectiveDelayMs, resumeAt }],
231+
}));
232+
nodeStates[String(node.id)] = { status: 'success', ts: completedTs };
233+
return { ticketPath: curTicketPath, lane: curLane, status: 'completed' };
234+
}
164235

165236
if (kind === 'llm') {
166237
const agentId = String(node?.assignedTo?.agentId ?? '');

‎src/lib/workflows/workflow-runner.ts‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ export * from './workflow-node-executor';
2222
export * from './workflow-worker';
2323
export * from './workflow-tick';
2424
export * from './workflow-approvals';
25+
export * from './workflow-if';
2526

2627
export async function enqueueWorkflowRun(api: OpenClawPluginApi, opts: {
2728
teamId: string;
@@ -159,7 +160,13 @@ export async function runWorkflowRunnerOnce(api: OpenClawPluginApi, opts: {
159160

160161
try {
161162
const run = JSON.parse(await readTextFile(runPath)) as RunLog;
162-
if (run.status !== 'queued') continue;
163+
const st = String(run.status ?? '');
164+
if (st !== 'queued' && st !== 'paused') continue;
165+
if (st === 'paused') {
166+
const resumeAtRaw = String(run.resumeAt ?? '').trim();
167+
const resumeMs = resumeAtRaw ? Date.parse(resumeAtRaw) : NaN;
168+
if (!Number.isFinite(resumeMs) || Date.now() < resumeMs) continue;
169+
}
163170
const exp = run.claimExpiresAt ? Date.parse(String(run.claimExpiresAt)) : 0;
164171
const claimed = !!run.claimedBy && exp > now;
165172
if (claimed) continue;

‎src/lib/workflows/workflow-tick.ts‎

Lines changed: 18 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -54,7 +54,16 @@ export async function runWorkflowRunnerTick(api: OpenClawPluginApi, opts: {
5454

5555
try {
5656
const run = JSON.parse(await readTextFile(runPath)) as RunLog;
57-
if (run.status !== 'queued') continue;
57+
const st = String(run.status ?? '');
58+
if (st !== 'queued' && st !== 'paused') continue;
59+
60+
// Paused runs (delay node): only resume once resumeAt has passed.
61+
if (st === 'paused') {
62+
const resumeAtRaw = String(run.resumeAt ?? '').trim();
63+
const resumeMs = resumeAtRaw ? Date.parse(resumeAtRaw) : NaN;
64+
if (!Number.isFinite(resumeMs) || Date.now() < resumeMs) continue;
65+
}
66+
5867
const exp = run.claimExpiresAt ? Date.parse(String(run.claimExpiresAt)) : 0;
5968
const claimed = !!run.claimedBy && exp > now;
6069
if (claimed) continue;
@@ -80,7 +89,13 @@ export async function runWorkflowRunnerTick(api: OpenClawPluginApi, opts: {
8089
async function tryClaim(runPath: string): Promise<RunLog | null> {
8190
const raw = await readTextFile(runPath);
8291
const cur = JSON.parse(raw) as RunLog;
83-
if (cur.status !== 'queued') return null;
92+
const st = String(cur.status ?? '');
93+
if (st !== 'queued' && st !== 'paused') return null;
94+
if (st === 'paused') {
95+
const resumeAtRaw = String(cur.resumeAt ?? '').trim();
96+
const resumeMs = resumeAtRaw ? Date.parse(resumeAtRaw) : NaN;
97+
if (!Number.isFinite(resumeMs) || Date.now() < resumeMs) return null;
98+
}
8499
const exp = cur.claimExpiresAt ? Date.parse(String(cur.claimExpiresAt)) : 0;
85100
const claimed = !!cur.claimedBy && exp > Date.now();
86101
if (claimed) return null;
@@ -92,6 +107,7 @@ export async function runWorkflowRunnerTick(api: OpenClawPluginApi, opts: {
92107
...cur,
93108
updatedAt: new Date().toISOString(),
94109
status: 'running',
110+
resumeAt: null,
95111
claimedBy,
96112
claimExpiresAt,
97113
events: [...(cur.events ?? []), { ts: new Date().toISOString(), type: 'run.claimed', claimedBy, claimExpiresAt }],

‎src/lib/workflows/workflow-types.ts‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,8 @@
11
export type WorkflowLane = 'backlog' | 'in-progress' | 'testing' | 'done';
22

3-
export type WorkflowNodeKind = 'llm' | 'human_approval' | 'writeback' | 'tool' | 'start' | 'end' | string;
3+
export type WorkflowNodeKind = 'llm' | 'human_approval' | 'writeback' | 'tool' | 'start' | 'end' | 'if' | 'delay' | string;
44

5-
export type WorkflowEdgeOn = 'success' | 'error' | 'always';
5+
export type WorkflowEdgeOn = 'success' | 'error' | 'always' | 'true' | 'false';
66

77
export type WorkflowNodeAssignment = {
88
agentId: string;
@@ -86,6 +86,8 @@ export type RunLog = {
8686
ticket: { file: string; number: string; lane: WorkflowLane };
8787
trigger: { kind: string; at?: string };
8888
status: string;
89+
// Delay/pause support (v1)
90+
resumeAt?: string | null;
8991
// Scheduler/runner fields
9092
priority?: number;
9193
claimedBy?: string | null;

0 commit comments

Comments
 (0)