diff --git a/ROADMAP.md b/ROADMAP.md index cb633c1491..83b0192383 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -148,7 +148,7 @@ Multi-stage triggers, action pipelines, execution logs, and cron scheduling stan | State machine & approval processes | ✅ | `automation/state-machine.zod.ts`, `automation/workflow.zod.ts` | | Retry policies with exponential backoff | ✅ | `automation/webhook.zod.ts` | | `IAutomationService` contract | ✅ | `contracts/automation-service.ts` (typed: `FlowParsed`, `ExecutionLog`) | -| `service-automation` DAG engine (MVP) | ✅ | `@objectstack/service-automation` (42 tests) | +| `service-automation` DAG engine (MVP) | ✅ | `@objectstack/service-automation` (67 tests) | | Execution log/history storage protocol | ✅ | `automation/execution.zod.ts` → `ExecutionLogSchema`, `ExecutionStepLogSchema` | | Execution error tracking & diagnostics | ✅ | `automation/execution.zod.ts` → `ExecutionErrorSchema`, `ExecutionErrorSeverity` | | Conflict resolution for concurrent executions | ✅ | `automation/execution.zod.ts` → `ConcurrencyPolicySchema` | @@ -157,6 +157,15 @@ Multi-stage triggers, action pipelines, execution logs, and cron scheduling stan | Automation API protocol (REST CRUD schemas) | ✅ | `api/automation-api.zod.ts` → 9 endpoints, all with `input`/`output` schemas | | Automation HTTP route handler (9 routes) | ✅ | `runtime/http-dispatcher.ts` → `handleAutomation()` CRUD + toggle + runs | | Client SDK `automation` namespace (10 methods) | ✅ | `client/src/index.ts` → `list`, `get`, `create`, `update`, `delete`, `toggle`, `runs.*` | +| Fault edge error path support | ✅ | `@objectstack/service-automation` → fault-type edge routing in DAG executor | +| Node step-level execution logging | ✅ | `@objectstack/service-automation` → per-node timing/status in `ExecutionLogEntry.steps` | +| Retry with exponential backoff & jitter | ✅ | `automation/flow.zod.ts` → `backoffMultiplier`, `maxRetryDelayMs`, `jitter` | +| Parallel branch execution (Promise.all) | ✅ | `@objectstack/service-automation` → unconditional edges run in parallel | +| Node timeout mechanism (Promise.race) | ✅ | `automation/flow.zod.ts` → `timeoutMs` per node, engine enforces via `Promise.race` | +| DAG cycle detection on registerFlow | ✅ | `@objectstack/service-automation` → DFS-based cycle detection with friendly error messages | +| Safe expression evaluation (no `new Function`) | ✅ | `@objectstack/service-automation` → operator-based parser, no code execution | +| Node input/output schema validation | ✅ | `automation/flow.zod.ts` → `inputSchema`/`outputSchema` per node, runtime validation | +| Flow version history & rollback | ✅ | `automation/flow.zod.ts` → `FlowVersionHistorySchema`, engine version management | ### 3. File Direct Upload & Resumable Upload Protocol @@ -446,13 +455,13 @@ business/custom objects, aligning with industry best practices (e.g., ServiceNow | Contract | Priority | Package | Notes | |:---|:---:|:---|:---| -| `IAutomationService` | **P2** | `@objectstack/service-automation` | ✅ Plugin-based DAG flow engine + HTTP API + Client SDK (42 tests) | +| `IAutomationService` | **P2** | `@objectstack/service-automation` | ✅ Plugin-based DAG flow engine + HTTP API + Client SDK (67 tests) | | `IWorkflowService` | **P2** | `@objectstack/service-workflow` | State machine + approval processes | | `IGraphQLService` | **P2** | `@objectstack/service-graphql` | Auto-generated GraphQL from objects | | `IAIService` | **P2** | `@objectstack/service-ai` | LLM integration (OpenAI/Anthropic/local) | | `IAnalyticsService` | **P3** | `@objectstack/service-analytics` | BI/OLAP queries | -- [x] `service-automation` — Implement `IAutomationService` with plugin-based DAG flow engine (MVP: CRUD/Logic/HTTP nodes), HTTP API CRUD (9 routes), Client SDK (10 methods), execution history +- [x] `service-automation` — Implement `IAutomationService` with plugin-based DAG flow engine (CRUD/Logic/HTTP nodes, fault edges, parallel branches, cycle detection, safe eval, timeout, versioning), HTTP API CRUD (9 routes), Client SDK (10 methods), execution history with step-level logging - [ ] `service-workflow` — Implement `IWorkflowService` with state machine runtime - [ ] `service-graphql` — Implement `IGraphQLService` with auto-schema generation - [ ] `service-ai` — Implement `IAIService` with multi-provider LLM routing @@ -721,7 +730,7 @@ Final polish and advanced features. | 16 | Search Service | `ISearchService` | ❌ | `@objectstack/service-search` (planned) | Spec only | | 17 | Notification Service | `INotificationService` | ❌ | `@objectstack/service-notification` (planned) | Spec only | | 18 | AI Service | `IAIService` | ❌ | `@objectstack/service-ai` (planned) | Spec only | -| 19 | Automation Service | `IAutomationService` | ✅ | `@objectstack/service-automation` | DAG engine + HTTP API CRUD + Client SDK + typed returns (42 tests) | +| 19 | Automation Service | `IAutomationService` | ✅ | `@objectstack/service-automation` | DAG engine + HTTP API CRUD + Client SDK + typed returns (67 tests) | | 20 | Workflow Service | `IWorkflowService` | ❌ | `@objectstack/service-workflow` (planned) | Spec only | | 21 | GraphQL Service | `IGraphQLService` | ❌ | `@objectstack/service-graphql` (planned) | Spec only | | 22 | i18n Service | `II18nService` | ✅ | `@objectstack/service-i18n` | File-based locale loading | diff --git a/packages/services/service-automation/src/engine.test.ts b/packages/services/service-automation/src/engine.test.ts index a2caf7289c..795c7ff2ec 100644 --- a/packages/services/service-automation/src/engine.test.ts +++ b/packages/services/service-automation/src/engine.test.ts @@ -787,3 +787,639 @@ describe('AutomationEngine - Execution History', () => { }); }); }); + +// ─── Fault Edge Tests ──────────────────────────────────────────────── + +describe('AutomationEngine - Fault Edge Support', () => { + let engine: AutomationEngine; + + beforeEach(() => { + engine = new AutomationEngine(createTestLogger()); + }); + + it('should follow fault edge when node fails', async () => { + const executed: string[] = []; + + engine.registerNodeExecutor({ + type: 'script', + async execute(node) { + if (node.id === 'risky') { + return { success: false, error: 'Script crashed' }; + } + executed.push(node.id); + return { success: true }; + }, + }); + + engine.registerFlow('fault_flow', { + name: 'fault_flow', + label: 'Fault Flow', + type: 'autolaunched', + variables: [{ name: 'status', type: 'text', isOutput: true }], + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'risky', type: 'script', label: 'Risky' }, + { id: 'handler', type: 'script', label: 'Error Handler' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'risky' }, + { id: 'e2', source: 'risky', target: 'end' }, + { id: 'e_fault', source: 'risky', target: 'handler', type: 'fault' }, + { id: 'e3', source: 'handler', target: 'end' }, + ], + }); + + const result = await engine.execute('fault_flow'); + expect(result.success).toBe(true); + expect(executed).toContain('handler'); + }); + + it('should write error info to $error variable on fault path', async () => { + let capturedError: unknown; + + engine.registerNodeExecutor({ + type: 'script', + async execute(node, variables) { + if (node.id === 'risky') { + return { success: false, error: 'Something went wrong' }; + } + capturedError = variables.get('$error'); + return { success: true }; + }, + }); + + engine.registerFlow('fault_error_ctx', { + name: 'fault_error_ctx', + label: 'Fault Error Context', + type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'risky', type: 'script', label: 'Risky' }, + { id: 'handler', type: 'script', label: 'Handler' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'risky' }, + { id: 'e2', source: 'risky', target: 'end' }, + { id: 'e_fault', source: 'risky', target: 'handler', type: 'fault' }, + { id: 'e3', source: 'handler', target: 'end' }, + ], + }); + + await engine.execute('fault_error_ctx'); + expect(capturedError).toBeDefined(); + expect((capturedError as any).message).toBe('Something went wrong'); + }); + + it('should throw when no fault edge and node fails', async () => { + engine.registerNodeExecutor({ + type: 'script', + async execute() { + return { success: false, error: 'Fatal error' }; + }, + }); + + engine.registerFlow('no_fault', { + name: 'no_fault', + label: 'No Fault', + type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'fail', type: 'script', label: 'Fail' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'fail' }, + { id: 'e2', source: 'fail', target: 'end' }, + ], + }); + + const result = await engine.execute('no_fault'); + expect(result.success).toBe(false); + expect(result.error).toContain('Fatal error'); + }); +}); + +// ─── Step-Level Execution Log Tests ────────────────────────────────── + +describe('AutomationEngine - Step-Level Execution Logs', () => { + let engine: AutomationEngine; + + beforeEach(() => { + engine = new AutomationEngine(createTestLogger()); + }); + + it('should record step logs with timing for each node', async () => { + engine.registerNodeExecutor({ + type: 'assignment', + async execute(node, variables) { + const config = (node.config ?? {}) as Record; + for (const [key, value] of Object.entries(config)) { + variables.set(key, value); + } + return { success: true }; + }, + }); + + engine.registerFlow('step_log_flow', { + name: 'step_log_flow', + label: 'Step Log Flow', + type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'assign', type: 'assignment', label: 'Assign', config: { x: 1 } }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'assign' }, + { id: 'e2', source: 'assign', target: 'end' }, + ], + }); + + await engine.execute('step_log_flow'); + const runs = await engine.listRuns('step_log_flow'); + expect(runs).toHaveLength(1); + expect(runs[0].steps.length).toBeGreaterThanOrEqual(2); // start + assign + expect(runs[0].steps[0].status).toBe('success'); + expect(runs[0].steps[0].startedAt).toBeTruthy(); + expect(typeof runs[0].steps[0].durationMs).toBe('number'); + }); + + it('should record failure step in logs when node fails', async () => { + engine.registerNodeExecutor({ + type: 'script', + async execute() { + return { success: false, error: 'Bad script' }; + }, + }); + + engine.registerFlow('fail_step_log', { + name: 'fail_step_log', + label: 'Fail Step Log', + type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'bad', type: 'script', label: 'Bad' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'bad' }, + { id: 'e2', source: 'bad', target: 'end' }, + ], + }); + + await engine.execute('fail_step_log'); + const runs = await engine.listRuns('fail_step_log'); + expect(runs).toHaveLength(1); + const failStep = runs[0].steps.find(s => s.nodeId === 'bad'); + expect(failStep).toBeDefined(); + expect(failStep!.status).toBe('failure'); + expect(failStep!.error).toBeDefined(); + }); + + it('should record flowVersion in execution log', async () => { + engine.registerFlow('versioned_flow', { + name: 'versioned_flow', + label: 'Versioned', + type: 'autolaunched', + version: 5, + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [{ id: 'e1', source: 'start', target: 'end' }], + }); + + await engine.execute('versioned_flow'); + const runs = await engine.listRuns('versioned_flow'); + expect(runs[0].flowVersion).toBe(5); + }); +}); + +// ─── DAG Cycle Detection Tests ─────────────────────────────────────── + +describe('AutomationEngine - DAG Cycle Detection', () => { + let engine: AutomationEngine; + + beforeEach(() => { + engine = new AutomationEngine(createTestLogger()); + }); + + it('should reject flows with cycles', () => { + expect(() => engine.registerFlow('cyclic_flow', { + name: 'cyclic_flow', + label: 'Cyclic Flow', + type: 'autolaunched', + nodes: [ + { id: 'a', type: 'start', label: 'A' }, + { id: 'b', type: 'assignment', label: 'B' }, + { id: 'c', type: 'assignment', label: 'C' }, + ], + edges: [ + { id: 'e1', source: 'a', target: 'b' }, + { id: 'e2', source: 'b', target: 'c' }, + { id: 'e3', source: 'c', target: 'b' }, // cycle: b → c → b + ], + })).toThrow(/cycle/i); + }); + + it('should accept valid DAG flows', () => { + expect(() => engine.registerFlow('valid_dag', { + name: 'valid_dag', + label: 'Valid DAG', + type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'a', type: 'assignment', label: 'A' }, + { id: 'b', type: 'assignment', label: 'B' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'a' }, + { id: 'e2', source: 'start', target: 'b' }, + { id: 'e3', source: 'a', target: 'end' }, + { id: 'e4', source: 'b', target: 'end' }, + ], + })).not.toThrow(); + }); + + it('should provide cycle details in error message', () => { + try { + engine.registerFlow('detailed_cycle', { + name: 'detailed_cycle', + label: 'Detailed Cycle', + type: 'autolaunched', + nodes: [ + { id: 'x', type: 'start', label: 'X' }, + { id: 'y', type: 'assignment', label: 'Y' }, + { id: 'z', type: 'assignment', label: 'Z' }, + ], + edges: [ + { id: 'e1', source: 'x', target: 'y' }, + { id: 'e2', source: 'y', target: 'z' }, + { id: 'e3', source: 'z', target: 'y' }, + ], + }); + expect.fail('Should have thrown'); + } catch (err: any) { + expect(err.message).toContain('→'); + expect(err.message).toContain('DAG'); + } + }); +}); + +// ─── Node Timeout Tests ────────────────────────────────────────────── + +describe('AutomationEngine - Node Timeout', () => { + let engine: AutomationEngine; + + beforeEach(() => { + engine = new AutomationEngine(createTestLogger()); + }); + + it('should timeout a slow node', async () => { + engine.registerNodeExecutor({ + type: 'script', + async execute() { + await new Promise(r => setTimeout(r, 5000)); // 5 seconds + return { success: true }; + }, + }); + + engine.registerFlow('timeout_flow', { + name: 'timeout_flow', + label: 'Timeout Flow', + type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'slow', type: 'script', label: 'Slow', timeoutMs: 50 }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'slow' }, + { id: 'e2', source: 'slow', target: 'end' }, + ], + }); + + const result = await engine.execute('timeout_flow'); + expect(result.success).toBe(false); + expect(result.error).toContain('timed out'); + }); + + it('should succeed when node completes within timeout', async () => { + engine.registerNodeExecutor({ + type: 'script', + async execute() { + return { success: true }; + }, + }); + + engine.registerFlow('fast_flow', { + name: 'fast_flow', + label: 'Fast Flow', + type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'fast', type: 'script', label: 'Fast', timeoutMs: 5000 }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'fast' }, + { id: 'e2', source: 'fast', target: 'end' }, + ], + }); + + const result = await engine.execute('fast_flow'); + expect(result.success).toBe(true); + }); +}); + +// ─── Safe Expression Evaluation Tests ──────────────────────────────── + +describe('AutomationEngine - Safe Expression Evaluation', () => { + let engine: AutomationEngine; + + beforeEach(() => { + engine = new AutomationEngine(createTestLogger()); + }); + + it('should evaluate simple comparisons', () => { + const vars = new Map(); + vars.set('amount', 500); + + expect(engine.evaluateCondition('{amount} > 100', vars)).toBe(true); + expect(engine.evaluateCondition('{amount} < 100', vars)).toBe(false); + expect(engine.evaluateCondition('{amount} == 500', vars)).toBe(true); + expect(engine.evaluateCondition('{amount} >= 500', vars)).toBe(true); + expect(engine.evaluateCondition('{amount} <= 500', vars)).toBe(true); + expect(engine.evaluateCondition('{amount} != 100', vars)).toBe(true); + }); + + it('should evaluate boolean literals', () => { + const vars = new Map(); + expect(engine.evaluateCondition('true', vars)).toBe(true); + expect(engine.evaluateCondition('false', vars)).toBe(false); + }); + + it('should not execute malicious code', () => { + const vars = new Map(); + // These should all return false safely + expect(engine.evaluateCondition('process.exit(1)', vars)).toBe(false); + expect(engine.evaluateCondition('require("fs").readFileSync("/etc/passwd")', vars)).toBe(false); + expect(engine.evaluateCondition('(() => { while(true) {} })()', vars)).toBe(false); + }); + + it('should handle string comparisons', () => { + const vars = new Map(); + vars.set('status', 'active'); + + expect(engine.evaluateCondition('{status} == active', vars)).toBe(true); + expect(engine.evaluateCondition('{status} != inactive', vars)).toBe(true); + }); +}); + +// ─── Parallel Branch Execution Tests ───────────────────────────────── + +describe('AutomationEngine - Parallel Branch Execution', () => { + let engine: AutomationEngine; + + beforeEach(() => { + engine = new AutomationEngine(createTestLogger()); + }); + + it('should execute unconditional branches in parallel', async () => { + const executionOrder: string[] = []; + + engine.registerNodeExecutor({ + type: 'script', + async execute(node) { + const delay = (node.config as any)?.delay ?? 0; + await new Promise(r => setTimeout(r, delay)); + executionOrder.push(node.id); + return { success: true }; + }, + }); + + engine.registerFlow('parallel_flow', { + name: 'parallel_flow', + label: 'Parallel Flow', + type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'branch_a', type: 'script', label: 'Branch A', config: { delay: 10 } }, + { id: 'branch_b', type: 'script', label: 'Branch B', config: { delay: 10 } }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'branch_a' }, + { id: 'e2', source: 'start', target: 'branch_b' }, + { id: 'e3', source: 'branch_a', target: 'end' }, + { id: 'e4', source: 'branch_b', target: 'end' }, + ], + }); + + const start = Date.now(); + const result = await engine.execute('parallel_flow'); + const elapsed = Date.now() - start; + + expect(result.success).toBe(true); + // Both branches should execute (order may vary in parallel) + expect(executionOrder).toContain('branch_a'); + expect(executionOrder).toContain('branch_b'); + // Parallel execution should be faster than sequential (10+10=20ms) + // Allow generous margin but expect it's faster than fully sequential + expect(elapsed).toBeLessThan(100); // generous but parallel should be ~15ms + }); +}); + +// ─── Input Schema Validation Tests ─────────────────────────────────── + +describe('AutomationEngine - Node Input Schema Validation', () => { + let engine: AutomationEngine; + + beforeEach(() => { + engine = new AutomationEngine(createTestLogger()); + }); + + it('should fail when required input parameter is missing', async () => { + engine.registerNodeExecutor({ + type: 'script', + async execute() { + return { success: true }; + }, + }); + + engine.registerFlow('schema_fail', { + name: 'schema_fail', + label: 'Schema Fail', + type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { + id: 'validated', + type: 'script', + label: 'Validated', + config: {}, + inputSchema: { + url: { type: 'string', required: true, description: 'URL to call' }, + }, + }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'validated' }, + { id: 'e2', source: 'validated', target: 'end' }, + ], + }); + + const result = await engine.execute('schema_fail'); + expect(result.success).toBe(false); + expect(result.error).toContain('missing required'); + }); + + it('should fail when parameter type is wrong', async () => { + engine.registerNodeExecutor({ + type: 'script', + async execute() { + return { success: true }; + }, + }); + + engine.registerFlow('type_fail', { + name: 'type_fail', + label: 'Type Fail', + type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { + id: 'validated', + type: 'script', + label: 'Validated', + config: { count: 'not_a_number' }, + inputSchema: { + count: { type: 'number', required: true }, + }, + }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'validated' }, + { id: 'e2', source: 'validated', target: 'end' }, + ], + }); + + const result = await engine.execute('type_fail'); + expect(result.success).toBe(false); + expect(result.error).toContain('expected type'); + }); +}); + +// ─── Flow Version Management Tests ─────────────────────────────────── + +describe('AutomationEngine - Flow Version Management', () => { + let engine: AutomationEngine; + + const makeFlow = (version: number, label: string) => ({ + name: 'versioned_flow', + label, + type: 'autolaunched' as const, + version, + nodes: [ + { id: 'start', type: 'start' as const, label: 'Start' }, + { id: 'end', type: 'end' as const, label: 'End' }, + ], + edges: [{ id: 'e1', source: 'start', target: 'end' }], + }); + + beforeEach(() => { + engine = new AutomationEngine(createTestLogger()); + }); + + it('should keep version history on registerFlow', () => { + engine.registerFlow('versioned_flow', makeFlow(1, 'V1')); + engine.registerFlow('versioned_flow', makeFlow(2, 'V2')); + engine.registerFlow('versioned_flow', makeFlow(3, 'V3')); + + const history = engine.getFlowVersionHistory('versioned_flow'); + expect(history).toHaveLength(3); + expect(history[0].version).toBe(1); + expect(history[2].version).toBe(3); + }); + + it('should rollback to a previous version', async () => { + engine.registerFlow('versioned_flow', makeFlow(1, 'V1')); + engine.registerFlow('versioned_flow', makeFlow(2, 'V2')); + + const current = await engine.getFlow('versioned_flow'); + expect(current!.label).toBe('V2'); + + engine.rollbackFlow('versioned_flow', 1); + const rolledBack = await engine.getFlow('versioned_flow'); + expect(rolledBack!.label).toBe('V1'); + }); + + it('should throw when rolling back to non-existent version', () => { + engine.registerFlow('versioned_flow', makeFlow(1, 'V1')); + expect(() => engine.rollbackFlow('versioned_flow', 99)).toThrow('Version 99 not found'); + }); + + it('should throw when rolling back non-existent flow', () => { + expect(() => engine.rollbackFlow('nonexistent', 1)).toThrow('no version history'); + }); + + it('should clean up version history on unregister', () => { + engine.registerFlow('versioned_flow', makeFlow(1, 'V1')); + engine.unregisterFlow('versioned_flow'); + const history = engine.getFlowVersionHistory('versioned_flow'); + expect(history).toHaveLength(0); + }); +}); + +// ─── Execution Status Expansion Tests ──────────────────────────────── + +describe('AutomationEngine - Execution Status', () => { + let engine: AutomationEngine; + + beforeEach(() => { + engine = new AutomationEngine(createTestLogger()); + }); + + it('should record completed status for successful execution', async () => { + engine.registerFlow('status_flow', { + name: 'status_flow', + label: 'Status Flow', + type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [{ id: 'e1', source: 'start', target: 'end' }], + }); + + await engine.execute('status_flow'); + const runs = await engine.listRuns('status_flow'); + expect(runs[0].status).toBe('completed'); + }); + + it('should record failed status for failed execution', async () => { + engine.registerFlow('fail_status', { + name: 'fail_status', + label: 'Fail Status', + type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'bad', type: 'script', label: 'Bad' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'bad' }, + { id: 'e2', source: 'bad', target: 'end' }, + ], + }); + + await engine.execute('fail_status'); + const runs = await engine.listRuns('fail_status'); + expect(runs[0].status).toBe('failed'); + }); +}); diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index 0549d80f4a..97a1fc6935 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -1,6 +1,7 @@ // Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. -import type { FlowParsed, FlowNodeParsed } from '@objectstack/spec/automation'; +import type { FlowParsed, FlowNodeParsed, FlowEdgeParsed } from '@objectstack/spec/automation'; +import type { ExecutionLog } from '@objectstack/spec/automation'; import type { AutomationContext, AutomationResult, IAutomationService } from '@objectstack/spec/contracts'; import type { Logger } from '@objectstack/spec/contracts'; import { FlowSchema } from '@objectstack/spec/automation'; @@ -52,23 +53,33 @@ export interface FlowTrigger { // ─── Core Automation Engine ───────────────────────────────────────── /** - * Internal execution log entry. + * Internal execution step log entry. + */ +interface StepLogEntry { + nodeId: string; + nodeType: string; + nodeLabel?: string; + status: 'success' | 'failure' | 'skipped'; + startedAt: string; + completedAt?: string; + durationMs?: number; + error?: { code: string; message: string; stack?: string }; +} + +/** + * Internal execution log entry — compatible with ExecutionLog from spec. */ interface ExecutionLogEntry { id: string; flowName: string; - status: 'completed' | 'failed'; + flowVersion?: number; + status: ExecutionLog['status']; startedAt: string; - completedAt: string; - durationMs: number; - trigger: { type: string; userId?: string; object?: string }; - steps: Array<{ - nodeId: string; - nodeType: string; - status: 'success' | 'failure' | 'skipped'; - startedAt: string; - durationMs?: number; - }>; + completedAt?: string; + durationMs?: number; + trigger: { type: string; userId?: string; object?: string; recordId?: string }; + steps: StepLogEntry[]; + variables?: Record; output?: unknown; error?: string; } @@ -76,6 +87,7 @@ interface ExecutionLogEntry { export class AutomationEngine implements IAutomationService { private flows = new Map(); private flowEnabled = new Map(); + private flowVersionHistory = new Map>(); private nodeExecutors = new Map(); private triggers = new Map(); private executionLogs: ExecutionLogEntry[] = []; @@ -130,16 +142,30 @@ export class AutomationEngine implements IAutomationService { registerFlow(name: string, definition: unknown): void { const parsed = FlowSchema.parse(definition); + + // DAG cycle detection + this.detectCycles(parsed); + + // 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); + this.flows.set(name, parsed); if (!this.flowEnabled.has(name)) { this.flowEnabled.set(name, true); } - this.logger.info(`Flow registered: ${name}`); + this.logger.info(`Flow registered: ${name} (version ${parsed.version})`); } unregisterFlow(name: string): void { this.flows.delete(name); this.flowEnabled.delete(name); + this.flowVersionHistory.delete(name); this.logger.info(`Flow unregistered: ${name}`); } @@ -159,6 +185,25 @@ export class AutomationEngine implements IAutomationService { this.logger.info(`Flow '${name}' ${enabled ? 'enabled' : 'disabled'}`); } + /** Get flow version history */ + getFlowVersionHistory(name: string): Array<{ version: number; definition: FlowParsed; createdAt: string }> { + return this.flowVersionHistory.get(name) ?? []; + } + + /** Rollback flow to a specific version */ + rollbackFlow(name: string, version: number): void { + const history = this.flowVersionHistory.get(name); + if (!history) { + throw new Error(`Flow '${name}' has no version history`); + } + const entry = history.find(h => h.version === version); + if (!entry) { + throw new Error(`Version ${version} not found for flow '${name}'`); + } + this.flows.set(name, entry.definition); + this.logger.info(`Flow '${name}' rolled back to version ${version}`); + } + async listRuns(flowName: string, options?: { limit?: number; cursor?: string }): Promise { const limit = options?.limit ?? 20; const logs = this.executionLogs.filter(l => l.flowName === flowName); @@ -198,6 +243,7 @@ export class AutomationEngine implements IAutomationService { const runId = `run_${++this.runCounter}`; const startedAt = new Date().toISOString(); + const steps: StepLogEntry[] = []; try { // Find the start node @@ -206,8 +252,11 @@ export class AutomationEngine implements IAutomationService { return { success: false, error: 'Flow has no start node' }; } + // Validate node input schemas before execution + this.validateNodeInputSchemas(flow, variables); + // DAG traversal execution - await this.executeNode(startNode, flow, variables, context ?? {}); + await this.executeNode(startNode, flow, variables, context ?? {}, steps); // Collect output variables const output: Record = {}; @@ -225,6 +274,7 @@ export class AutomationEngine implements IAutomationService { this.recordLog({ id: runId, flowName, + flowVersion: flow.version, status: 'completed', startedAt, completedAt: new Date().toISOString(), @@ -234,7 +284,7 @@ export class AutomationEngine implements IAutomationService { userId: context?.userId, object: context?.object, }, - steps: [], + steps, output, }); @@ -251,6 +301,7 @@ export class AutomationEngine implements IAutomationService { this.recordLog({ id: runId, flowName, + flowVersion: flow.version, status: 'failed', startedAt, completedAt: new Date().toISOString(), @@ -260,7 +311,7 @@ export class AutomationEngine implements IAutomationService { userId: context?.userId, object: context?.object, }, - steps: [], + steps, error: errorMessage, }); @@ -286,27 +337,212 @@ export class AutomationEngine implements IAutomationService { } } + /** + * Detect cycles in the flow graph (DAG validation). + * Uses DFS with coloring (white/gray/black) to detect back edges. + * Throws an error with cycle details if a cycle is found. + */ + private detectCycles(flow: FlowParsed): void { + const WHITE = 0, GRAY = 1, BLACK = 2; + const color = new Map(); + const parent = new Map(); + + // Build adjacency list from edges + const adj = new Map(); + for (const node of flow.nodes) { + color.set(node.id, WHITE); + adj.set(node.id, []); + } + for (const edge of flow.edges) { + const targets = adj.get(edge.source); + if (targets) targets.push(edge.target); + } + + const dfs = (nodeId: string): string[] | null => { + color.set(nodeId, GRAY); + for (const neighbor of adj.get(nodeId) ?? []) { + if (color.get(neighbor) === GRAY) { + // Back edge found — reconstruct cycle + const cycle = [neighbor, nodeId]; + let cur = nodeId; + while (cur !== neighbor) { + cur = parent.get(cur)!; + if (cur) cycle.push(cur); + else break; + } + return cycle.reverse(); + } + if (color.get(neighbor) === WHITE) { + parent.set(neighbor, nodeId); + const result = dfs(neighbor); + if (result) return result; + } + } + color.set(nodeId, BLACK); + return null; + }; + + for (const node of flow.nodes) { + if (color.get(node.id) === WHITE) { + const cycle = dfs(node.id); + if (cycle) { + throw new Error(`Flow contains a cycle: ${cycle.join(' → ')}. Only DAG flows are allowed.`); + } + } + } + } + + /** + * Get the runtime type name of a value for schema validation. + */ + private getValueType(value: unknown): string { + if (Array.isArray(value)) return 'array'; + if (typeof value === 'object' && value !== null) return 'object'; + return typeof value; + } + + /** + * Validate node input schemas before execution. + * Checks that node config matches declared inputSchema if present. + */ + private validateNodeInputSchemas(flow: FlowParsed, _variables: Map): void { + for (const node of flow.nodes) { + if (node.inputSchema && node.config) { + for (const [paramName, paramDef] of Object.entries(node.inputSchema)) { + if (paramDef.required && !(paramName in (node.config as Record))) { + throw new Error( + `Node '${node.id}' missing required input parameter '${paramName}'`, + ); + } + const value = (node.config as Record)[paramName]; + if (value !== undefined) { + const actualType = this.getValueType(value); + if (actualType !== paramDef.type) { + throw new Error( + `Node '${node.id}' parameter '${paramName}' expected type '${paramDef.type}' but got '${actualType}'`, + ); + } + } + } + } + } + } + + /** + * Execute a node with timeout support, fault edge handling, and step logging. + */ private async executeNode( node: FlowNodeParsed, flow: FlowParsed, variables: Map, context: AutomationContext, + steps: StepLogEntry[], ): Promise { if (node.type === 'end') return; + const stepStart = Date.now(); + const stepStartedAt = new Date().toISOString(); + // Find executor const executor = this.nodeExecutors.get(node.type); if (!executor) { // start node without executor is fine — just skip if (node.type !== 'start') { + steps.push({ + nodeId: node.id, + nodeType: node.type, + status: 'failure', + startedAt: stepStartedAt, + completedAt: new Date().toISOString(), + durationMs: Date.now() - stepStart, + error: { code: 'NO_EXECUTOR', message: `No executor registered for node type '${node.type}'` }, + }); throw new Error(`No executor registered for node type '${node.type}'`); } + // Log start node step + steps.push({ + nodeId: node.id, + nodeType: node.type, + status: 'success', + startedAt: stepStartedAt, + completedAt: new Date().toISOString(), + durationMs: Date.now() - stepStart, + }); } else { - // Execute node - const result = await executor.execute(node, variables, context); + // Execute node with optional timeout + let result: NodeExecutionResult; + try { + if (node.timeoutMs && node.timeoutMs > 0) { + result = await this.executeWithTimeout( + executor.execute(node, variables, context), + node.timeoutMs, + node.id, + ); + } else { + result = await executor.execute(node, variables, context); + } + } catch (execErr: unknown) { + const errMsg = execErr instanceof Error ? execErr.message : String(execErr); + steps.push({ + nodeId: node.id, + nodeType: node.type, + status: 'failure', + startedAt: stepStartedAt, + completedAt: new Date().toISOString(), + durationMs: Date.now() - stepStart, + error: { code: 'EXECUTION_ERROR', message: errMsg }, + }); + + // Check for fault edges + const faultEdge = flow.edges.find(e => e.source === node.id && e.type === 'fault'); + if (faultEdge) { + variables.set('$error', { nodeId: node.id, message: errMsg }); + const faultTarget = flow.nodes.find(n => n.id === faultEdge.target); + if (faultTarget) { + await this.executeNode(faultTarget, flow, variables, context, steps); + return; + } + } + throw execErr; + } + if (!result.success) { - throw new Error(`Node '${node.id}' failed: ${result.error}`); + const errMsg = result.error ?? 'Unknown error'; + steps.push({ + nodeId: node.id, + nodeType: node.type, + status: 'failure', + startedAt: stepStartedAt, + completedAt: new Date().toISOString(), + durationMs: Date.now() - stepStart, + error: { code: 'NODE_FAILURE', message: errMsg }, + }); + + // Write error output to variable context for downstream nodes + variables.set('$error', { nodeId: node.id, message: errMsg, output: result.output }); + + // Check for fault edges + const faultEdge = flow.edges.find(e => e.source === node.id && e.type === 'fault'); + if (faultEdge) { + const faultTarget = flow.nodes.find(n => n.id === faultEdge.target); + if (faultTarget) { + await this.executeNode(faultTarget, flow, variables, context, steps); + return; + } + } + throw new Error(`Node '${node.id}' failed: ${errMsg}`); } + + // Log successful step + steps.push({ + nodeId: node.id, + nodeType: node.type, + status: 'success', + startedAt: stepStartedAt, + completedAt: new Date().toISOString(), + durationMs: Date.now() - stepStart, + }); + // Write back output variables if (result.output) { for (const [key, value] of Object.entries(result.output)) { @@ -315,48 +551,257 @@ export class AutomationEngine implements IAutomationService { } } - // Find next nodes (filter by edge conditions) - const outEdges = flow.edges.filter(e => e.source === node.id); + // Find next nodes — separate conditional and unconditional edges + const outEdges = flow.edges.filter( + e => e.source === node.id && e.type !== 'fault', + ); + + const conditionalEdges: FlowEdgeParsed[] = []; + const unconditionalEdges: FlowEdgeParsed[] = []; for (const edge of outEdges) { - if (edge.condition && !this.evaluateCondition(edge.condition, variables)) { - continue; + if (edge.condition) { + conditionalEdges.push(edge); + } else { + unconditionalEdges.push(edge); } - const nextNode = flow.nodes.find(n => n.id === edge.target); - if (nextNode) { - await this.executeNode(nextNode, flow, variables, context); + } + + // Conditional edges: evaluate sequentially (mutually exclusive) + for (const edge of conditionalEdges) { + if (this.evaluateCondition(edge.condition!, variables)) { + const nextNode = flow.nodes.find(n => n.id === edge.target); + if (nextNode) { + await this.executeNode(nextNode, flow, variables, context, steps); + } } } + + // Unconditional edges: execute in parallel (Promise.all) + if (unconditionalEdges.length > 0) { + 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); + } + } + + /** + * Execute a promise with timeout using Promise.race. + */ + private executeWithTimeout( + promise: Promise, + timeoutMs: number, + nodeId: string, + ): Promise { + return Promise.race([ + promise, + new Promise((_, reject) => + setTimeout(() => reject(new Error(`Node '${nodeId}' timed out after ${timeoutMs}ms`)), timeoutMs), + ), + ]); } - private evaluateCondition(expression: string, variables: Map): boolean { - // MVP: Simple template replacement + expression evaluation. - // Flow definitions are authored by trusted developers/admins. - // TODO: Replace with safe expression evaluator (e.g., jexl) for production. + /** + * Safe expression evaluator. + * Uses simple operator-based parsing without `new Function`. + * Supports: comparisons (>, <, >=, <=, ==, !=, ===, !==), + * boolean literals (true, false), and basic arithmetic. + */ + evaluateCondition(expression: string, variables: Map): boolean { + // Template replacement: {varName} → value let resolved = expression; for (const [key, value] of variables) { resolved = resolved.split(`{${key}}`).join(String(value)); } + resolved = resolved.trim(); + try { - return new Function(`return (${resolved})`)() as boolean; + // Boolean literals + if (resolved === 'true') return true; + if (resolved === 'false') return false; + + // Comparison operators (ordered by length to match longer operators first) + const operators = ['===', '!==', '>=', '<=', '!=', '==', '>', '<'] as const; + for (const op of operators) { + const idx = resolved.indexOf(op); + if (idx !== -1) { + const left = resolved.slice(0, idx).trim(); + const right = resolved.slice(idx + op.length).trim(); + return this.compareValues(left, op, right); + } + } + + // Numeric truthy check + const numVal = Number(resolved); + if (!isNaN(numVal)) return numVal !== 0; + + return false; } catch { return false; } } + /** + * Compare two string-represented values with an operator. + */ + private compareValues(left: string, op: string, right: string): boolean { + const lNum = Number(left); + const rNum = Number(right); + const bothNumeric = !isNaN(lNum) && !isNaN(rNum) && left !== '' && right !== ''; + + if (bothNumeric) { + switch (op) { + case '>': return lNum > rNum; + case '<': return lNum < rNum; + case '>=': return lNum >= rNum; + case '<=': return lNum <= rNum; + case '==': case '===': return lNum === rNum; + case '!=': case '!==': return lNum !== rNum; + default: return false; + } + } + // String comparison + switch (op) { + case '==': case '===': return left === right; + case '!=': case '!==': return left !== right; + case '>': return left > right; + case '<': return left < right; + case '>=': return left >= right; + case '<=': return left <= right; + default: return false; + } + } + + /** + * Retry execution with exponential backoff, jitter, and recursive protection. + * Uses an iterative loop with an internal retry flag to prevent recursive call stacking. + */ private async retryExecution( flowName: string, context: AutomationContext | undefined, startTime: number, - errorHandling: { maxRetries?: number; retryDelayMs?: number }, + errorHandling: { + maxRetries?: number; + retryDelayMs?: number; + backoffMultiplier?: number; + maxRetryDelayMs?: number; + jitter?: boolean; + }, ): Promise { 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 }; + } + + /** + * Execute a flow without triggering retry logic (used by retryExecution to prevent recursion). + */ + private async executeWithoutRetry( + flowName: string, + context?: AutomationContext, + ): Promise { + const startTime = Date.now(); + const flow = this.flows.get(flowName); + + if (!flow) { + return { success: false, error: `Flow '${flowName}' not found` }; + } + if (this.flowEnabled.get(flowName) === false) { + return { success: false, error: `Flow '${flowName}' is disabled` }; + } + + const variables = new Map(); + if (flow.variables) { + for (const v of flow.variables) { + if (v.isInput && context?.params?.[v.name] !== undefined) { + variables.set(v.name, context.params[v.name]); + } + } + } + if (context?.record) { + variables.set('$record', context.record); + } + + const runId = `run_${++this.runCounter}`; + const startedAt = new Date().toISOString(); + const steps: StepLogEntry[] = []; + + 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); + + const output: Record = {}; + if (flow.variables) { + for (const v of flow.variables) { + if (v.isOutput) { + output[v.name] = variables.get(v.name); + } + } + } + + const durationMs = Date.now() - startTime; + this.recordLog({ + id: runId, + flowName, + flowVersion: flow.version, + status: 'completed', + startedAt, + completedAt: new Date().toISOString(), + durationMs, + trigger: { + type: context?.event ?? 'manual', + userId: context?.userId, + object: context?.object, + }, + steps, + output, + }); + + return { success: true, output, durationMs }; + } catch (err: unknown) { + const errorMessage = err instanceof Error ? err.message : String(err); + const durationMs = Date.now() - startTime; + this.recordLog({ + id: runId, + flowName, + flowVersion: flow.version, + status: 'failed', + startedAt, + completedAt: new Date().toISOString(), + durationMs, + trigger: { + type: context?.event ?? 'manual', + userId: context?.userId, + object: context?.object, + }, + steps, + error: errorMessage, + }); + return { success: false, error: errorMessage, durationMs }; } - return { success: false, error: 'Max retries exceeded', durationMs: Date.now() - startTime }; } } diff --git a/packages/services/service-automation/src/plugins/logic-nodes-plugin.ts b/packages/services/service-automation/src/plugins/logic-nodes-plugin.ts index e36bfbbb9b..8b739d3eea 100644 --- a/packages/services/service-automation/src/plugins/logic-nodes-plugin.ts +++ b/packages/services/service-automation/src/plugins/logic-nodes-plugin.ts @@ -25,19 +25,8 @@ export class LogicNodesPlugin implements Plugin { const conditions = (config?.conditions ?? []) as Array<{ label: string; expression: string }>; for (const cond of conditions) { - // MVP: Simple template replacement + expression evaluation. - // Flow definitions are authored by trusted developers/admins. - // TODO: Replace with safe expression evaluator (e.g., jexl) for production. - let expr = cond.expression; - for (const [k, v] of variables) { - expr = expr.split(`{${k}}`).join(String(v)); - } - try { - if (new Function(`return (${expr})`)()) { - return { success: true, branchLabel: cond.label }; - } - } catch { - // Continue to next condition + if (engine.evaluateCondition(cond.expression, variables)) { + return { success: true, branchLabel: cond.label }; } } return { success: true, branchLabel: 'default' }; diff --git a/packages/spec/src/automation/flow.test.ts b/packages/spec/src/automation/flow.test.ts index a1e3c54b53..6b8d5432cf 100644 --- a/packages/spec/src/automation/flow.test.ts +++ b/packages/spec/src/automation/flow.test.ts @@ -5,6 +5,7 @@ import { FlowEdgeSchema, FlowVariableSchema, FlowNodeAction, + FlowVersionHistorySchema, defineFlow, type Flow, type FlowNode, @@ -130,6 +131,39 @@ describe('FlowNodeSchema', () => { expect(() => FlowNodeSchema.parse(node)).not.toThrow(); }); }); + + it('should accept node with timeoutMs', () => { + const result = FlowNodeSchema.safeParse({ + id: 'http_1', + type: 'http_request', + label: 'Call API', + timeoutMs: 5000, + }); + expect(result.success).toBe(true); + if (result.success) { + expect(result.data.timeoutMs).toBe(5000); + } + }); + + it('should accept node with inputSchema and outputSchema', () => { + const result = FlowNodeSchema.safeParse({ + id: 'script_1', + type: 'script', + label: 'Process Data', + inputSchema: { + name: { type: 'string', required: true, description: 'User name' }, + age: { type: 'number', required: false }, + }, + outputSchema: { + greeting: { type: 'string', description: 'Generated greeting' }, + }, + }); + expect(result.success).toBe(true); + if (result.success) { + expect(result.data.inputSchema).toBeDefined(); + expect(result.data.outputSchema).toBeDefined(); + } + }); }); describe('FlowEdgeSchema', () => { @@ -597,6 +631,53 @@ describe('FlowSchema - errorHandling', () => { }); expect(result.errorHandling).toBeUndefined(); }); + + it('should accept exponential backoff configuration', () => { + const result = FlowSchema.safeParse({ + name: 'backoff_flow', + label: 'Backoff Flow', + type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [{ id: 'e1', source: 'start', target: 'end' }], + errorHandling: { + strategy: 'retry', + maxRetries: 5, + retryDelayMs: 1000, + backoffMultiplier: 2, + maxRetryDelayMs: 30000, + jitter: true, + }, + }); + expect(result.success).toBe(true); + if (result.success) { + expect(result.data.errorHandling!.backoffMultiplier).toBe(2); + expect(result.data.errorHandling!.maxRetryDelayMs).toBe(30000); + expect(result.data.errorHandling!.jitter).toBe(true); + } + }); + + it('should use defaults for backoff fields', () => { + const result = FlowSchema.safeParse({ + name: 'default_backoff', + label: 'Default Backoff', + type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [{ id: 'e1', source: 'start', target: 'end' }], + errorHandling: { strategy: 'retry' }, + }); + expect(result.success).toBe(true); + if (result.success) { + expect(result.data.errorHandling!.backoffMultiplier).toBe(1); + expect(result.data.errorHandling!.maxRetryDelayMs).toBe(30000); + expect(result.data.errorHandling!.jitter).toBe(false); + } + }); }); describe('defineFlow', () => { @@ -642,3 +723,31 @@ describe('defineFlow', () => { })).toThrow(); }); }); + +describe('FlowVersionHistorySchema', () => { + it('should validate a flow version history entry', () => { + const result = FlowVersionHistorySchema.safeParse({ + flowName: 'my_flow', + version: 1, + definition: { + name: 'my_flow', + label: 'My Flow', + type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [{ id: 'e1', source: 'start', target: 'end' }], + }, + createdAt: '2026-01-01T00:00:00Z', + createdBy: 'admin', + changeNote: 'Initial version', + }); + expect(result.success).toBe(true); + }); + + it('should require flowName, version, definition, and createdAt', () => { + const result = FlowVersionHistorySchema.safeParse({}); + expect(result.success).toBe(false); + }); +}); diff --git a/packages/spec/src/automation/flow.zod.ts b/packages/spec/src/automation/flow.zod.ts index 06a5b3eff9..8ee9910e63 100644 --- a/packages/spec/src/automation/flow.zod.ts +++ b/packages/spec/src/automation/flow.zod.ts @@ -72,6 +72,22 @@ export const FlowNodeSchema = z.object({ /** UI Position (for the canvas) */ position: z.object({ x: z.number(), y: z.number() }).optional(), + + /** Node-level execution timeout */ + timeoutMs: z.number().int().min(0).optional().describe('Maximum execution time for this node in milliseconds'), + + /** Node input schema declaration for Studio form generation and runtime validation */ + inputSchema: z.record(z.string(), z.object({ + type: z.enum(['string', 'number', 'boolean', 'object', 'array']).describe('Parameter type'), + required: z.boolean().default(false).describe('Whether the parameter is required'), + description: z.string().optional().describe('Parameter description'), + })).optional().describe('Input parameter schema for this node'), + + /** Node output schema declaration */ + outputSchema: z.record(z.string(), z.object({ + type: z.enum(['string', 'number', 'boolean', 'object', 'array']).describe('Output type'), + description: z.string().optional().describe('Output description'), + })).optional().describe('Output schema declaration for this node'), }); /** @@ -143,6 +159,9 @@ export const FlowSchema = z.object({ strategy: z.enum(['fail', 'retry', 'continue']).default('fail').describe('How to handle node execution errors'), maxRetries: z.number().int().min(0).max(10).default(0).describe('Number of retry attempts (only for retry strategy)'), retryDelayMs: z.number().int().min(0).default(1000).describe('Delay between retries in milliseconds'), + backoffMultiplier: z.number().min(1).default(1).describe('Multiplier for exponential backoff between retries'), + maxRetryDelayMs: z.number().int().min(0).default(30000).describe('Maximum delay between retries in milliseconds'), + jitter: z.boolean().default(false).describe('Add random jitter to retry delay to avoid thundering herd'), fallbackNodeId: z.string().optional().describe('Node ID to jump to on unrecoverable error'), }).optional().describe('Flow-level error handling configuration'), }); @@ -176,3 +195,21 @@ export type FlowNode = z.input; export type FlowNodeParsed = z.infer; export type FlowEdge = z.input; export type FlowEdgeParsed = z.infer; + +/** + * Flow Version History Schema + * Tracks historical versions of flow definitions for rollback support. + * + * Industry alignment: Salesforce Flow Versions, n8n Workflow History. + */ +export const FlowVersionHistorySchema = z.object({ + flowName: z.string().describe('Flow machine name'), + version: z.number().int().min(1).describe('Version number'), + definition: FlowSchema.describe('Complete flow definition snapshot'), + createdAt: z.string().datetime().describe('When this version was created'), + createdBy: z.string().optional().describe('User who created this version'), + changeNote: z.string().optional().describe('Description of what changed in this version'), +}); + +export type FlowVersionHistory = z.input; +export type FlowVersionHistoryParsed = z.infer;