tests: add unit tests for DAG builder

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

@@ -233,3 +233,4 @@ export const WorkflowYamlSchema = z.object({
 });
 
 export type WorkflowYaml = z.infer<typeof WorkflowYamlSchema>;
+export type WorkflowSchema = z.infer<typeof WorkflowSchema>;

src/platform/packages/shared/kbn-workflows/types/execution/index.ts

@@ -16,3 +16,4 @@ export type {
   ExitConditionBranchNodeSchema,
 } from './nodes/branching_nodes';
 export type { EnterForeachNode, ExitForeachNode } from './nodes/loop_nodes';
+export type { AtomicGraphNode } from './nodes/base';

src/platform/packages/shared/kbn-workflows/types/execution/nodes/base.ts

@@ -16,3 +16,10 @@ export const ExecutionGraphNodeSchema = z.object({
 });
 
 export type ExecutionGraphNode = z.infer<typeof ExecutionGraphNodeSchema>;
+
+export const AtomicGraphNodeSchema = z.object({
+  id: z.string(),
+  type: z.literal('atomic'),
+  configuration: z.any(),
+});
+export type AtomicGraphNode = z.infer<typeof AtomicGraphNodeSchema>;

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

@@ -52,4 +52,5 @@ export type {
   ExitConditionBranchNodeSchema,
   EnterForeachNode,
   ExitForeachNode,
+  AtomicGraphNode,
 } from './execution';

src/platform/plugins/shared/workflows_execution_engine/server/step/atomic_step/atomic_step_impl.ts

@@ -0,0 +1,51 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+import { AtomicGraphNode } from '@kbn/workflows/types/execution/nodes/base';
+import { StepImplementation } from '../step_base';
+import { ConnectorStepImpl } from '../connector_step';
+import { WorkflowContextManager } from '../../workflow_context_manager/workflow_context_manager';
+import { ConnectorExecutor } from '../../connector_executor';
+import { WorkflowExecutionRuntimeManager } from '../../workflow_context_manager/workflow_execution_runtime_manager';
+
+/**
+ * Implements the execution logic for an atomic workflow step.
+ *
+ * `AtomicStepImpl` is responsible for running a single atomic step within a workflow.
+ * It delegates the execution to a `ConnectorStepImpl`, passing the necessary configuration,
+ * context manager, connector executor, and workflow state.
+ *
+ * @remarks
+ * This class is typically used internally by the workflow execution engine to process
+ * atomic nodes in the workflow graph.
+ *
+ * @param node - The atomic graph node containing step configuration.
+ * @param contextManager - Manages workflow context and state.
+ * @param connectorExecutor - Executes connector operations for the step.
+ * @param workflowState - Manages the runtime state of workflow execution.
+ */
+export class AtomicStepImpl implements StepImplementation {
+  constructor(
+    private node: AtomicGraphNode,
+    private contextManager: WorkflowContextManager,
+    private connectorExecutor: ConnectorExecutor,
+    private workflowState: WorkflowExecutionRuntimeManager
+  ) {}
+
+  async run(): Promise<void> {
+    // This class should decide what action to take based on action type
+    // like connector, logger, http call, etc.
+    // for now it only calls ConnectorStepImpl
+    await new ConnectorStepImpl(
+      this.node.configuration,
+      this.contextManager,
+      this.connectorExecutor,
+      this.workflowState
+    ).run();
+  }
+}

src/platform/plugins/shared/workflows_execution_engine/server/step/step_factory.ts

@@ -12,7 +12,6 @@ import { WorkflowContextManager } from '../workflow_context_manager/workflow_con
 import { StepImplementation } from './step_base';
 // Import schema and inferred types
 import { ConnectorExecutor } from '../connector_executor';
-import { ConnectorStepImpl } from './connector_step';
 import {
   EnterConditionBranchNodeImpl,
   EnterIfNodeImpl,
@@ -21,6 +20,7 @@ import {
 } from './if_step';
 import { WorkflowExecutionRuntimeManager } from '../workflow_context_manager/workflow_execution_runtime_manager';
 import { EnterForeachNodeImpl, ExitForeachNodeImpl } from './foreach_step';
+import { AtomicStepImpl } from './atomic_step/atomic_step_impl';
 // Import specific step implementations
 // import { ForEachStepImpl } from './foreach-step'; // To be created
 // import { IfStepImpl } from './if-step'; // To be created
@@ -30,7 +30,7 @@ import { EnterForeachNodeImpl, ExitForeachNodeImpl } from './foreach_step';
 
 export class StepFactory {
   public create<TStep extends BaseStep>(
-    step: TStep, // Use z.infer<typeof StepSchema> when fully defined
+    step: TStep, // TODO: TStep must refer to a node type, not BaseStep (IfElseNode, ForeachNode, etc.)
     contextManager: WorkflowContextManager,
     connectorExecutor: ConnectorExecutor, // this is temporary, we will remove it when we have a proper connector executor
     workflowState: WorkflowExecutionRuntimeManager
@@ -55,13 +55,13 @@ export class StepFactory {
       case 'exit-if':
         return new ExitIfNodeImpl(step as any, workflowState);
       case 'atomic':
-      // return new AtomicStepImpl(step as AtomicStep, contextManager);
+        return new AtomicStepImpl(step as any, contextManager, connectorExecutor, workflowState);
       case 'parallel':
       // return new ParallelStepImpl(step as ParallelStep, contextManager);
       case 'merge':
       // return new MergeStepImpl(step as MergeStep, contextManager);
       default:
-        return new ConnectorStepImpl(step as any, contextManager, connectorExecutor, workflowState);
+        throw new Error(`Unknown node type: ${stepType}`);
     }
   }
 }