Uh oh!
There was an error while loading. Please reload this page.
Automation engine: fault edges, safe eval, cycle detection, parallel execution, timeouts, versioning - #770
Conversation
The latest updates on your projects. Learn more about Vercel for GitHub.
|
…ion history to flow spec - Add backoffMultiplier, maxRetryDelayMs, jitter to FlowSchema.errorHandling - Add timeoutMs, inputSchema, outputSchema to FlowNodeSchema - Add FlowVersionHistorySchema for rollback support - Add comprehensive tests for all new fields Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
…ut, cycle detection, safe eval, parallel execution, versioning, and backoff - Add backoffMultiplier, maxRetryDelayMs, jitter to FlowSchema.errorHandling - Add timeoutMs, inputSchema, outputSchema to FlowNodeSchema - Add FlowVersionHistorySchema for flow version management - Implement fault edge error path support in DAG executor - Add node step-level execution logging with timing - Implement retry with exponential backoff, jitter, and recursive protection - Add parallel branch execution (Promise.all for unconditional edges) - Add node timeout mechanism (Promise.race) - Implement DAG cycle detection on registerFlow - Replace new Function with safe expression evaluator - Add node input/output schema validation - Add flow version history management and rollback Co-authored-by: hotlong <50353452+hotlong@users.noreply.github.com>
…and update ROADMAP - Add tests for fault edge support (3 tests) - Add tests for step-level execution logs (3 tests) - Add tests for DAG cycle detection (3 tests) - Add tests for node timeout mechanism (2 tests) - Add tests for safe expression evaluation (4 tests) - Add tests for parallel branch execution (1 test) - Add tests for input schema validation (2 tests) - Add tests for flow version management (5 tests) - Add tests for execution status recording (2 tests) - Update ROADMAP.md with all new automation features Co-authored-by: hotlong <50353452+hotlong@users.noreply.github.com>
… simplify string comparison switch Co-authored-by: hotlong <50353452+hotlong@users.noreply.github.com>
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Pull request overview
This PR expands the automation flow spec and the service-automation engine to support more production-grade DAG execution features (fault routing, safe condition evaluation, timeouts, retries/backoff, parallel branches, cycle detection, and version history), aligned with common automation platforms.
Changes:
- Extended
FlowNodeSchemaandFlowSchema.errorHandlingwith timeout, I/O schema declarations, and exponential backoff configuration; addedFlowVersionHistorySchema. - Refactored the automation engine to add cycle detection, fault-edge routing, step-level execution logs, safe condition evaluation (no
new Function), parallel branch execution, node timeouts, and version history/rollback. - Added/updated unit tests for the new spec fields and engine behaviors; updated roadmap counts and feature checklist.
Reviewed changes
Copilot reviewed 6 out of 6 changed files in this pull request and generated 10 comments.
Show a summary per file
| File | Description |
|---|---|
| packages/spec/src/automation/flow.zod.ts | Adds node timeout + I/O schema declarations, retry backoff fields, and a flow version history schema. |
| packages/spec/src/automation/flow.test.ts | Adds tests covering the new spec fields/schemas. |
| packages/services/service-automation/src/plugins/logic-nodes-plugin.ts | Switches decision-node condition evaluation to the engine’s safe evaluator (removes new Function). |
| packages/services/service-automation/src/engine.ts | Implements the new engine capabilities: step logs, fault edges, cycle detection, safe eval, parallelism, timeouts, retries/backoff, input validation, and version history/rollback. |
| packages/services/service-automation/src/engine.test.ts | Adds broad test coverage for the newly introduced engine behaviors. |
| ROADMAP.md | Updates automation engine test count and marks new engine capabilities as completed. |
| engine.registerNodeExecutor({ | ||
| type: 'script', | ||
| async execute() { | ||
| await new Promise(r => setTimeout(r, 5000)); // 5 seconds |
There was a problem hiding this comment.
The timeout test’s executor intentionally sleeps for 5 seconds. Since the underlying promise isn’t cancelled when the engine times out, that timer will still keep the Node event loop alive and can add ~5s to the overall test runtime. Consider using fake timers or reducing the sleep duration to something just above timeoutMs (e.g., 200ms) to keep the suite fast and reliable.
| awaitnewPromise(r=>setTimeout(r,5000));// 5 seconds | |
| awaitnewPromise(r=>setTimeout(r,200));// slightly above timeoutMs to simulate slowness |
| // Version history management | ||
| const history = this.flowVersionHistory.get(name) ?? []; | ||
| history.push({ | ||
| version: parsed.version, | ||
| definition: parsed, | ||
| createdAt: new Date().toISOString(), | ||
| }); | ||
| this.flowVersionHistory.set(name, history); |
There was a problem hiding this comment.
registerFlow() records version history using parsed.version, but FlowSchema defaults version to 1. If callers don’t explicitly bump version, you’ll store multiple history entries with the same version, and rollbackFlow() will pick the first match, making rollback ambiguous/incorrect. Consider auto-incrementing versions on registration or enforcing uniqueness/monotonicity per flow.
| * Execute a promise with timeout using Promise.race. | ||
| */ | ||
| private executeWithTimeout( | ||
| promise: Promise<NodeExecutionResult>, | ||
| timeoutMs: number, | ||
| nodeId: string, | ||
| ): Promise<NodeExecutionResult> { | ||
| return Promise.race([ | ||
| promise, | ||
| new Promise<NodeExecutionResult>((_, reject) => | ||
| setTimeout(() => reject(new Error(`Node '${nodeId}' timed out after ${timeoutMs}ms`)), timeoutMs), | ||
| ), | ||
| ]); |
There was a problem hiding this comment.
executeWithTimeout() creates a setTimeout that isn’t cleared when the main promise resolves first. This can keep timers alive unnecessarily under load. Consider storing the timeout handle and clearTimeout() it once the primary execution settles.
| *Executeapromisewithtimeout using Promise.race. | |
| */ | |
| privateexecuteWithTimeout( | |
| promise: Promise<NodeExecutionResult>, | |
| timeoutMs: number, | |
| nodeId: string, | |
| ): Promise<NodeExecutionResult>{ | |
| returnPromise.race([ | |
| promise, | |
| newPromise<NodeExecutionResult>((_,reject)=> | |
| setTimeout(()=>reject(newError(`Node '${nodeId}' timed out after ${timeoutMs}ms`)),timeoutMs), | |
| ), | |
| ]); | |
| *Executeapromisewithtimeout using acancellabletimer. | |
| */ | |
| privateexecuteWithTimeout( | |
| promise: Promise<NodeExecutionResult>, | |
| timeoutMs: number, | |
| nodeId: string, | |
| ): Promise<NodeExecutionResult>{ | |
| returnnewPromise<NodeExecutionResult>((resolve,reject)=>{ | |
| consttimeoutId=setTimeout( | |
| ()=>reject(newError(`Node '${nodeId}' timed out after ${timeoutMs}ms`)), | |
| timeoutMs, | |
| ); | |
| promise.then( | |
| result=>{ | |
| clearTimeout(timeoutId); | |
| resolve(result); | |
| }, | |
| error=>{ | |
| clearTimeout(timeoutId); | |
| reject(error); | |
| }, | |
| ); | |
| }); |
| * Safe expression evaluator. | ||
| * Uses simple operator-based parsing without `new Function`. | ||
| * Supports: comparisons (>, <, >=, <=, ==, !=, ===, !==), | ||
| * boolean literals (true, false), and basic arithmetic. |
There was a problem hiding this comment.
The evaluateCondition() docstring says it supports “basic arithmetic”, but the implementation only handles boolean literals, comparisons, and numeric truthy checks. This is misleading for future maintainers and consumers. Either implement arithmetic parsing or update the comment to match current capabilities.
| * boolean literals(true,false),andbasicarithmetic. | |
| * boolean literals(true,false),andnumerictruthinesschecks(non-zero→true). |
| try { | ||
| const startNode = flow.nodes.find(n => n.type === 'start'); | ||
| if (!startNode) { | ||
| return { success: false, error: 'Flow has no start node' }; | ||
| } | ||
| await this.executeNode(startNode, flow, variables, context ?? {}, steps); | ||
There was a problem hiding this comment.
Retry attempts via executeWithoutRetry() don’t perform the same preflight checks as execute() (notably validateNodeInputSchemas()). That makes behavior differ between the initial attempt and retries. Consider sharing a single internal execution path so retries apply the same validation/logging behavior consistently.
| * Industry alignment: Salesforce Flow Versions, n8n Workflow History. | ||
| */ | ||
| export const FlowVersionHistorySchema = z.object({ | ||
| flowName: z.string().describe('Flow machine name'), |
There was a problem hiding this comment.
FlowVersionHistorySchema.flowName is described as a “machine name”, but unlike FlowSchema.name it doesn’t enforce the repo’s snake_case machine-name regex. This makes version-history entries accept invalid flow names and diverges from existing conventions. Consider reusing the same regex (or the same schema) as FlowSchema.name.
| flowName: z.string().describe('Flow machine name'), | |
| flowName: FlowSchema.shape.name.describe('Flow machine name'), |
| const targets = adj.get(edge.source); | ||
| if (targets) targets.push(edge.target); |
There was a problem hiding this comment.
detectCycles() silently ignores edges whose source isn’t in the node list (adj.get(edge.source) returns undefined). That means invalid graphs can pass registration, and cycle detection can miss cycles involving typo’d node IDs. Consider validating that every edge’s source/target exists and throwing a descriptive error when they don’t.
| consttargets=adj.get(edge.source); | |
| if(targets)targets.push(edge.target); | |
| if(!adj.has(edge.source)){ | |
| thrownewError( | |
| `Invalid flow graph: edge references missing source node '${edge.source}'.`, | |
| ); | |
| } | |
| if(!adj.has(edge.target)){ | |
| thrownewError( | |
| `Invalid flow graph: edge references missing target node '${edge.target}'.`, | |
| ); | |
| } | |
| consttargets=adj.get(edge.source)!; | |
| targets.push(edge.target); |
| if (node.inputSchema && node.config) { | ||
| for (const [paramName, paramDef] of Object.entries(node.inputSchema)) { | ||
| if (paramDef.required && !(paramName in (node.config as Record<string, unknown>))) { | ||
| throw new Error( | ||
| `Node '${node.id}' missing required input parameter '${paramName}'`, | ||
| ); | ||
| } | ||
| const value = (node.config as Record<string, unknown>)[paramName]; |
There was a problem hiding this comment.
validateNodeInputSchemas() only runs when node.inputSchema && node.config. If inputSchema declares required params but config is missing/undefined, required parameters won’t be validated at all. Consider treating missing config as {} (and still validating required fields) whenever inputSchema is present.
| if(node.inputSchema&&node.config){ | |
| for(const[paramName,paramDef]ofObject.entries(node.inputSchema)){ | |
| if(paramDef.required&&!(paramNamein(node.configasRecord<string,unknown>))){ | |
| thrownewError( | |
| `Node '${node.id}' missing required input parameter '${paramName}'`, | |
| ); | |
| } | |
| constvalue=(node.configasRecord<string,unknown>)[paramName]; | |
| if(node.inputSchema){ | |
| constconfig=(node.config??{})asRecord<string,unknown>; | |
| for(const[paramName,paramDef]ofObject.entries(node.inputSchema)){ | |
| if(paramDef.required&&!(paramNameinconfig)){ | |
| thrownewError( | |
| `Node '${node.id}' missing required input parameter '${paramName}'`, | |
| ); | |
| } | |
| constvalue=config[paramName]; |
| const parallelTasks = unconditionalEdges | ||
| .map(edge => flow.nodes.find(n => n.id === edge.target)) | ||
| .filter((n): n is FlowNodeParsed => n != null) | ||
| .map(nextNode => this.executeNode(nextNode, flow, variables, context, steps)); | ||
| await Promise.all(parallelTasks); |
There was a problem hiding this comment.
Unconditional edges are executed in parallel but share the same mutable variables Map across branches. That creates a real race condition (last write wins) and can make executions nondeterministic when parallel nodes write to the same variables. Consider isolating variables per branch (clone/overlay then merge deterministically) or explicitly enforcing read-only variables in parallel branches.
| constparallelTasks=unconditionalEdges | |
| .map(edge=>flow.nodes.find(n=>n.id===edge.target)) | |
| .filter((n): n is FlowNodeParsed=>n!=null) | |
| .map(nextNode=>this.executeNode(nextNode,flow,variables,context,steps)); | |
| awaitPromise.all(parallelTasks); | |
| // Clone variables per branch to avoid parallel mutation of shared Map. | |
| constbranchVariableMaps: Map<string,unknown>[]=[]; | |
| constparallelTasks=unconditionalEdges | |
| .map(edge=>{ | |
| constnextNode=flow.nodes.find(n=>n.id===edge.target); | |
| if(!nextNode){ | |
| returnnull; | |
| } | |
| // Each branch gets its own variable context snapshot. | |
| constbranchVariables=newMap<string,unknown>(variables); | |
| branchVariableMaps.push(branchVariables); | |
| returnthis.executeNode(nextNode,flow,branchVariables,context,steps); | |
| }) | |
| .filter((task): task is Promise<void>=>task!=null); | |
| awaitPromise.all(parallelTasks); | |
| // Deterministically merge branch variables back into the shared context. | |
| for(constbranchVariablesofbranchVariableMaps){ | |
| for(const[key,value]ofbranchVariables){ | |
| variables.set(key,value); | |
| } | |
| } |
| const maxRetries = errorHandling.maxRetries ?? 3; | ||
| const delay = errorHandling.retryDelayMs ?? 1000; | ||
| const baseDelay = errorHandling.retryDelayMs ?? 1000; | ||
| const multiplier = errorHandling.backoffMultiplier ?? 1; | ||
| const maxDelay = errorHandling.maxRetryDelayMs ?? 30000; | ||
| const useJitter = errorHandling.jitter ?? false; | ||
| let lastError = 'Max retries exceeded'; | ||
| for (let i = 0; i < maxRetries; i++) { | ||
| // Calculate delay with exponential backoff | ||
| let delay = Math.min(baseDelay * Math.pow(multiplier, i), maxDelay); | ||
| if (useJitter) { | ||
| delay = delay * (0.5 + Math.random() * 0.5); | ||
| } | ||
| await new Promise(r => setTimeout(r, delay)); | ||
| const result = await this.execute(flowName, context); | ||
| // Execute directly without recursion into retryExecution again | ||
| const result = await this.executeWithoutRetry(flowName, context); | ||
| if (result.success) return result; | ||
| lastError = result.error ?? 'Unknown error'; | ||
| } | ||
| return { success: false, error: lastError, durationMs: Date.now() - startTime }; | ||
| } |
There was a problem hiding this comment.
retryExecution() treats maxRetries = 0 as “no loop iterations” and returns 'Max retries exceeded', which can mask the original failure reason. Since FlowSchema.errorHandling.maxRetries defaults to 0, a strategy: 'retry' flow can end up reporting the wrong error. Consider returning the original error when maxRetries is 0, and/or interpreting maxRetries as “number of retries after the initial attempt”.
Deep refinement of the automation engine architecture, benchmarked against n8n/Activepieces/Zapier patterns. Covers 9 of 11 tasks from the tracking issue.
Spec Schema (
packages/spec/src/automation/flow.zod.ts)backoffMultiplier,maxRetryDelayMs,jitteronerrorHandlingtimeoutMsonFlowNodeSchemainputSchema/outputSchemaonFlowNodeSchemafor Studio form generation and runtime validationFlowVersionHistorySchemafor rollback supportEngine (
packages/services/service-automation/src/engine.ts)type: 'fault'edges instead of aborting. Error context written to$errorvariable for downstream nodes.startedAt,completedAt,durationMs,status,errorintosteps[]on the execution log.new Function()with an operator-based string parser. No code execution possible — verified with malicious input tests.registerFlow, rejects cyclic graphs with descriptive error.Promise.all; conditional edges evaluate sequentially.Promise.racewrapper whentimeoutMs > 0.retryExecutioncomputes delay asbaseDelay * multiplier^attemptcapped atmaxRetryDelayMs, optional jitter. UsesexecuteWithoutRetryto prevent recursive call stacking.inputSchema.registerFlowappends to version history;rollbackFlow(name, version)restores a snapshot.Tests
67 engine tests (25 new), 5996 spec tests (6 new). All pre-existing tests pass unchanged.
Security
Removed
new Function()from bothengine.tsandlogic-nodes-plugin.ts. The replacement parser only handles comparison operators and boolean literals — noeval-family calls remain.Original prompt
💡 You can make Copilot smarter by setting up custom instructions, customizing its development environment and configuring Model Context Protocol (MCP) servers. Learn more Copilot coding agent tips in the docs.