[One Workflow] fix: http step error handling lacks structured error data

src/platform/packages/shared/kbn-workflows/graph/build_execution_graph/build_execution_graph.ts

@@ -378,8 +378,13 @@ function visitOnFailure(
     graph = createFallback(stepId, graph, onFailureConfiguration.fallback, context);
   }
 
-  if (onFailureConfiguration.continue) {
-    graph = createContinue(stepId, graph);
+  // Here we can statically determine that 'continue' is needed if "continue" is boolean.
+  // If condition is a string (expression), we need to evaluate it at runtime, so we always create the continue node.
+  if (
+    typeof onFailureConfiguration.continue === 'string' ||
+    onFailureConfiguration.continue === true
+  ) {
+    graph = createContinue(stepId, onFailureConfiguration.continue, graph);
   }
 
   context.stack.pop();
@@ -470,7 +475,11 @@ function handleWorkflowLevelOnFailure(
   return result;
 }
 
-function createContinue(stepId: string, innerGraph: WorkflowGraphType): WorkflowGraphType {
+function createContinue(
+  stepId: string,
+  condition: string | boolean,
+  innerGraph: WorkflowGraphType
+): WorkflowGraphType {
   const graph = createTypedGraph({ directed: true });
   const enterContinueNodeId = `enterContinue_${stepId}`;
   const exitNodeId = `exitContinue_${stepId}`;
@@ -480,6 +489,9 @@ function createContinue(stepId: string, innerGraph: WorkflowGraphType): Workflow
     stepId,
     stepType: 'continue',
     exitNodeId,
+    configuration: {
+      condition,
+    },
   };
   const exitContinueNode: ExitContinueNode = {
     type: 'exit-continue',

src/platform/packages/shared/kbn-workflows/graph/types/nodes/on_failure_nodes.ts

@@ -14,7 +14,9 @@ import { WorkflowRetrySchema } from '../../../spec/schema';
 export const EnterContinueNodeSchema = GraphNodeSchema.extend({
   id: z.string(),
   type: z.literal('enter-continue'),
-
+  configuration: z.object({
+    condition: z.union([z.string(), z.boolean()]),
+  }),
   exitNodeId: z.string(),
 });
 export type EnterContinueNode = z.infer<typeof EnterContinueNodeSchema>;

src/platform/packages/shared/kbn-workflows/spec/schema.ts

@@ -20,6 +20,7 @@ export const RetryPolicySchema = z.object({
 
 export const WorkflowRetrySchema = z.object({
   'max-attempts': z.number().min(1),
+  condition: z.string().optional(), // e.g., "${{error.type == 'NetworkError'}}" (default: always retry)
   delay: z
     .string()
     .regex(/^\d+(ms|[smhdw])$/, 'Invalid duration format')
@@ -37,7 +38,7 @@ export type BaseStep = z.infer<typeof BaseStepSchema>;
 export const WorkflowOnFailureSchema = z.object({
   retry: WorkflowRetrySchema.optional(),
   fallback: z.array(BaseStepSchema).min(1).optional(),
-  continue: z.boolean().optional(),
+  continue: z.union([z.boolean(), z.string()]).optional(),
 });
 
 export type WorkflowOnFailure = z.infer<typeof WorkflowOnFailureSchema>;
@@ -589,3 +590,10 @@ export const DynamicStepContextSchema = DynamicWorkflowContextSchema.extend({
   steps: z.object({}),
 });
 export type DynamicStepContext = z.infer<typeof DynamicStepContextSchema>;
+
+export const ExecutionErrorSchema = z.object({
+  type: z.string(),
+  message: z.string(),
+  details: z.any().optional(),
+});
+export type ExecutionError = z.infer<typeof ExecutionErrorSchema>;

src/platform/packages/shared/kbn-workflows/types/v1.ts

@@ -9,7 +9,7 @@
 
 import type { JsonValue } from '@kbn/utility-types';
 import { z } from '@kbn/zod';
-import type { WorkflowYaml } from '../spec/schema';
+import type { ExecutionError, WorkflowYaml } from '../spec/schema';
 import { WorkflowSchema } from '../spec/schema';
 
 export enum ExecutionStatus {
@@ -74,7 +74,7 @@ export interface EsWorkflowExecution {
   stepId?: string;
   scopeStack: StackFrame[];
   createdAt: string;
-  error: string | null;
+  error: ExecutionError | null;
   createdBy: string;
   startedAt: string;
   finishedAt: string;
@@ -126,7 +126,7 @@ export interface EsWorkflowStepExecution {
    * There might be several instances of the same stepId if it's inside loops, retries, etc.
    */
   stepExecutionIndex: number;
-  error?: string | null;
+  error?: ExecutionError;
   output?: JsonValue;
   input?: JsonValue;
 

src/platform/plugins/shared/workflows_execution_engine/integration_tests/tests/on_failure_continue.test.ts

@@ -19,18 +19,15 @@ describe('workflow with continue on failure', () => {
   });
 
   describe.each(['step level', 'workflow level'])('continue is on %s', (testCase) => {
-    let buildYamlFn: () => string;
-
-    beforeAll(() => {
+    function buildYamlFn(continueExpression: boolean | string | undefined): string {
       if (testCase === 'step level') {
-        buildYamlFn = () => {
-          return `
+        return `
 steps:
   - name: constantlyFailingStep
     type: ${FakeConnectors.constantlyFailing.actionTypeId}
     connector-id: ${FakeConnectors.constantlyFailing.name}
-    on-failure:
-      continue: true
+    ${continueExpression ? 'on-failure:' : ''}
+      ${continueExpression ? `continue: ${continueExpression}` : ''}
     with:
       message: 'Hi there! Are you alive?'
 
@@ -40,13 +37,11 @@ steps:
     with:
       message: 'Final message!'
 `;
-        };
-      } else if (testCase === 'workflow level') {
-        buildYamlFn = () => {
-          return `
+      }
+      return `
 settings:
-  on-failure:
-    continue: true
+  ${continueExpression ? 'on-failure:' : ''}
+    ${continueExpression ? `continue: ${continueExpression}` : ''}
 steps:
   - name: constantlyFailingStep
     type: ${FakeConnectors.constantlyFailing.actionTypeId}
@@ -60,122 +55,122 @@ steps:
     with:
       message: 'Final message!'
 `;
-        };
-      }
-    });
+    }
+
+    describe.each([true, '${{error.type == "Error"}}'])(
+      'when continue is %s',
+      (continueTestCase) => {
+        beforeEach(async () => {
+          jest.clearAllMocks();
+          await workflowRunFixture.runWorkflow({
+            workflowYaml: buildYamlFn(continueTestCase),
+          });
+        });
 
-    describe('when continue is true', () => {
-      beforeEach(async () => {
-        jest.clearAllMocks();
-        await workflowRunFixture.runWorkflow({
-          workflowYaml: buildYamlFn(),
+        it('should successfully complete workflow despite error in constantlyFailingStep', async () => {
+          const workflowExecutionDoc =
+            workflowRunFixture.workflowExecutionRepositoryMock.workflowExecutions.get(
+              'fake_workflow_execution_id'
+            );
+          expect(workflowExecutionDoc?.status).toBe(ExecutionStatus.COMPLETED);
+          expect(workflowExecutionDoc?.error).toBe(undefined);
+          expect(workflowExecutionDoc?.scopeStack).toEqual([]);
         });
-      });
 
-      it('should successfully complete workflow despite error in constantlyFailingStep', async () => {
-        const workflowExecutionDoc =
-          workflowRunFixture.workflowExecutionRepositoryMock.workflowExecutions.get(
-            'fake_workflow_execution_id'
+        it('should execute constantlyFailingStep once and record its failure', async () => {
+          const failingStepExecutions = Array.from(
+            workflowRunFixture.stepExecutionRepositoryMock.stepExecutions.values()
+          ).filter(
+            (se) =>
+              se.stepId === 'constantlyFailingStep' &&
+              se.stepType === FakeConnectors.constantlyFailing.actionTypeId
           );
-        expect(workflowExecutionDoc?.status).toBe(ExecutionStatus.COMPLETED);
-        expect(workflowExecutionDoc?.error).toBe(undefined);
-        expect(workflowExecutionDoc?.scopeStack).toEqual([]);
-      });
-
-      it('should execute constantlyFailingStep once and record its failure', async () => {
-        const failingStepExecutions = Array.from(
-          workflowRunFixture.stepExecutionRepositoryMock.stepExecutions.values()
-        ).filter(
-          (se) =>
-            se.stepId === 'constantlyFailingStep' &&
-            se.stepType === FakeConnectors.constantlyFailing.actionTypeId
-        );
-        expect(failingStepExecutions.length).toBe(1);
-        expect(failingStepExecutions[0].status).toBe(ExecutionStatus.FAILED);
-        expect(failingStepExecutions[0].error).toBe('Error: Constantly failing connector');
-      });
-
-      it('should execute finalStep successfully after the failing step', async () => {
-        const finalStepExecutions = Array.from(
-          workflowRunFixture.stepExecutionRepositoryMock.stepExecutions.values()
-        ).filter((se) => se.stepId === 'finalStep');
-        expect(finalStepExecutions.length).toBe(1);
-        expect(finalStepExecutions[0].status).toBe(ExecutionStatus.COMPLETED);
-        expect(finalStepExecutions[0].error).toBe(undefined);
-      });
-
-      it('should execute finalStep after constantlyFailingStep', async () => {
-        const failingStepExecutions = Array.from(
-          workflowRunFixture.stepExecutionRepositoryMock.stepExecutions.values()
-        ).filter(
-          (se) =>
-            se.stepId === 'constantlyFailingStep' &&
-            se.stepType === FakeConnectors.constantlyFailing.actionTypeId
-        );
-        const finalStepExecutions = Array.from(
-          workflowRunFixture.stepExecutionRepositoryMock.stepExecutions.values()
-        ).filter((se) => se.stepId === 'finalStep');
-
-        expect(failingStepExecutions.length).toBe(1);
-        expect(finalStepExecutions.length).toBe(1);
-
-        const failingStepCompletedAt = new Date(failingStepExecutions[0].completedAt!).getTime();
-        const finalStepStartedAt = new Date(finalStepExecutions[0].startedAt).getTime();
-
-        expect(finalStepStartedAt).toBeGreaterThanOrEqual(failingStepCompletedAt);
-      });
-    });
-  });
+          expect(failingStepExecutions.length).toBe(1);
+          expect(failingStepExecutions[0].status).toBe(ExecutionStatus.FAILED);
+          expect(failingStepExecutions[0].error).toEqual({
+            message: 'Error: Constantly failing connector',
+            type: 'Error',
+          });
+        });
 
-  describe('when continue is false (default behavior)', () => {
-    beforeEach(async () => {
-      jest.clearAllMocks();
-      await workflowRunFixture.runWorkflow({
-        workflowYaml: `
-steps:
-  - name: constantlyFailingStep
-    type: ${FakeConnectors.constantlyFailing.actionTypeId}
-    connector-id: ${FakeConnectors.constantlyFailing.name}
-    with:
-      message: 'Hi there! Are you alive?'
+        it('should execute finalStep successfully after the failing step', async () => {
+          const finalStepExecutions = Array.from(
+            workflowRunFixture.stepExecutionRepositoryMock.stepExecutions.values()
+          ).filter((se) => se.stepId === 'finalStep');
+          expect(finalStepExecutions.length).toBe(1);
+          expect(finalStepExecutions[0].status).toBe(ExecutionStatus.COMPLETED);
+          expect(finalStepExecutions[0].error).toBe(undefined);
+        });
 
-  - name: finalStep
-    type: slack
-    connector-id: ${FakeConnectors.slack2.name}
-    with:
-      message: 'Final message!'
-`,
-      });
-    });
-
-    it('should fail the workflow due to error in constantlyFailingStep', async () => {
-      const workflowExecutionDoc =
-        workflowRunFixture.workflowExecutionRepositoryMock.workflowExecutions.get(
-          'fake_workflow_execution_id'
-        );
-      expect(workflowExecutionDoc?.status).toBe(ExecutionStatus.FAILED);
-      expect(workflowExecutionDoc?.error).toBe('Error: Constantly failing connector');
-      expect(workflowExecutionDoc?.scopeStack).toEqual([]);
-    });
-
-    it('should execute constantlyFailingStep once and record its failure', async () => {
-      const failingStepExecutions = Array.from(
-        workflowRunFixture.stepExecutionRepositoryMock.stepExecutions.values()
-      ).filter(
-        (se) =>
-          se.stepId === 'constantlyFailingStep' &&
-          se.stepType === FakeConnectors.constantlyFailing.actionTypeId
-      );
-      expect(failingStepExecutions.length).toBe(1);
-      expect(failingStepExecutions[0].status).toBe(ExecutionStatus.FAILED);
-      expect(failingStepExecutions[0].error).toBe('Error: Constantly failing connector');
-    });
-
-    it('should not execute finalStep', async () => {
-      const finalStepExecutions = Array.from(
-        workflowRunFixture.stepExecutionRepositoryMock.stepExecutions.values()
-      ).filter((se) => se.stepId === 'finalStep');
-      expect(finalStepExecutions.length).toBe(0);
-    });
+        it('should execute finalStep after constantlyFailingStep', async () => {
+          const failingStepExecutions = Array.from(
+            workflowRunFixture.stepExecutionRepositoryMock.stepExecutions.values()
+          ).filter(
+            (se) =>
+              se.stepId === 'constantlyFailingStep' &&
+              se.stepType === FakeConnectors.constantlyFailing.actionTypeId
+          );
+          const finalStepExecutions = Array.from(
+            workflowRunFixture.stepExecutionRepositoryMock.stepExecutions.values()
+          ).filter((se) => se.stepId === 'finalStep');
+
+          expect(failingStepExecutions.length).toBe(1);
+          expect(finalStepExecutions.length).toBe(1);
+
+          const failingStepCompletedAt = new Date(failingStepExecutions[0].completedAt!).getTime();
+          const finalStepStartedAt = new Date(finalStepExecutions[0].startedAt).getTime();
+
+          expect(finalStepStartedAt).toBeGreaterThanOrEqual(failingStepCompletedAt);
+        });
+      }
+    );
+
+    describe.each([undefined, false, '${{error.type == "SomeOtherError"}}'])(
+      'when continue is "%s"',
+      (continueTestCase) => {
+        beforeEach(async () => {
+          jest.clearAllMocks();
+          await workflowRunFixture.runWorkflow({
+            workflowYaml: buildYamlFn(continueTestCase),
+          });
+        });
+
+        it('should fail the workflow due to error in constantlyFailingStep', async () => {
+          const workflowExecutionDoc =
+            workflowRunFixture.workflowExecutionRepositoryMock.workflowExecutions.get(
+              'fake_workflow_execution_id'
+            );
+          expect(workflowExecutionDoc?.status).toBe(ExecutionStatus.FAILED);
+          expect(workflowExecutionDoc?.error).toEqual({
+            message: 'Error: Constantly failing connector',
+            type: 'Error',
+          });
+          expect(workflowExecutionDoc?.scopeStack).toEqual([]);
+        });
+
+        it('should execute constantlyFailingStep once and record its failure', async () => {
+          const failingStepExecutions = Array.from(
+            workflowRunFixture.stepExecutionRepositoryMock.stepExecutions.values()
+          ).filter(
+            (se) =>
+              se.stepId === 'constantlyFailingStep' &&
+              se.stepType === FakeConnectors.constantlyFailing.actionTypeId
+          );
+          expect(failingStepExecutions.length).toBe(1);
+          expect(failingStepExecutions[0].status).toBe(ExecutionStatus.FAILED);
+          expect(failingStepExecutions[0].error).toEqual({
+            message: 'Error: Constantly failing connector',
+            type: 'Error',
+          });
+        });
+
+        it('should not execute finalStep', async () => {
+          const finalStepExecutions = Array.from(
+            workflowRunFixture.stepExecutionRepositoryMock.stepExecutions.values()
+          ).filter((se) => se.stepId === 'finalStep');
+          expect(finalStepExecutions.length).toBe(0);
+        });
+      }
+    );
   });
 });