Repository: kestra-io/kestra
Issue: Create a Flowable task type Dag with dependsOn
Labels: backend
enhancement

### Discussed in https://github.com/kestra-io/kestra/discussions/1143

This kind of heavy dependent graph is complicated to do in Kestra: 
![SimpleGraph](https://g.gravizo.com/svg?%20digraph%20G%20{%20start%20-%3E%20a0;%20start%20-%3E%20b0;%20start%20-%3E%20c0;%20start%20-%3E%20d0;%20start%20-%3E%20e0;%20a0%20-%3E%20a1_0;%20a0%20-%3E%20a1_1;%20a0%20-%3E%20a1_2;%20a1_1%20-%3E%20a2_0;%20a1_1%20-%3E%20a2_1;%20b0%20-%3E%20bc1_0;%20c0%20-%3E%20bc1_0;%20bc1_0%20-%3E%20bc2;%20c0%20-%3E%20c1_1;%20c1_1%20-%3E%20e2;%20e0%20-%3E%20e1;%20e1%20-%3E%20e2;%20start%20[shape=Mdiamond];%20}%20))


My suggest is to create a [Flowable task](https://kestra.io/docs/concepts/flows#flowable-task) is will have this structure : 

```yaml
type: io.kestra.core.tasks.flows.Dag
tasks: 
  - task: 
      id: a 
      type: io.kestra.core.tasks.debugs.Return
  - task: 
      id: b 
      type: io.kestra.core.tasks.debugs.Return
  - task: 
      id: c
      type: io.kestra.core.tasks.debugs.Return
    dependsOn : 
      - a
      - b
  - task: 
      id: d
      type: io.kestra.core.tasks.debugs.Return
    dependsOn : 
      - c
```

That will allow more complex dag to be created with out the pain of build manually dependencies. 

Issue comments:
> [batuhanfaik](/batuhanfaik) [ May 15, 2023 ](https://github.com/kestra-io/kestra/discussions/1143#discussioncomment-5904698)
> LGTM, thanks a lot for opening this! Maybe we can add the code of whole solution to the sample tree as it involves most of the edge cases that people (including myself) face when solving DAGs :)

@batuhanfaik, didn't understand your comment? 

Sorry, I could've been more clear. Your suggestion looks good to me. I wanted to say that we can add the sample code which would actually solve the tree presented in the figure. However, with the introduction of simple "dependsOn" keyword, we don't need to write the whole thing. It's pretty self explanatory 

Pull request review comments:
default to `Runtime.availableProcessor() * 2` as it's a sane default (I have a PR somewhere to do the same for the Parallel tasks)
I'll rename TaskDepend to DagTask and move the class inside this one.
```suggestion
    @NotEmpty
```
return an empty list instead of null
you can return a list as it may be multiple cyclic dependencies
```suggestion
    private List<String> nestedDependencies(TaskDepend taskDepend, List<TaskDepend> tasks, List<String> visited) {
```
I think this can be an array list as you only use the list at index 0
Why do you check on last created ?
This is really hard to read
It's part of the parallel task on which the dag task is based

Patch that resolved the issue:
diff --git a/core/src/main/java/io/kestra/core/runners/FlowableUtils.java b/core/src/main/java/io/kestra/core/runners/FlowableUtils.java
index 03acf3c25b..f93fe7b6d9 100644
--- a/core/src/main/java/io/kestra/core/runners/FlowableUtils.java
+++ b/core/src/main/java/io/kestra/core/runners/FlowableUtils.java
@@ -11,8 +11,11 @@
 import io.kestra.core.models.tasks.ResolvedTask;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.serializers.JacksonMapper;
+import io.kestra.core.tasks.flows.Dag;
 
 import java.util.*;
+import java.util.function.BiFunction;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -134,13 +137,68 @@ public static List<ResolvedTask> resolveTasks(List<Task> tasks, TaskRun parentTa
             .collect(Collectors.toList());
     }
 
-
     public static List<NextTaskRun> resolveParallelNexts(
         Execution execution,
         List<ResolvedTask> tasks,
         List<ResolvedTask> errors,
         TaskRun parentTaskRun,
         Integer concurrency
+    ) {
+        return resolveParallelNexts(
+            execution,
+            tasks, errors,
+            parentTaskRun,
+            concurrency,
+            (nextTaskRunStream, taskRuns) -> nextTaskRunStream
+        );
+    }
+
+    public static List<NextTaskRun> resolveDagNexts(
+        Execution execution,
+        List<ResolvedTask> tasks,
+        List<ResolvedTask> errors,
+        TaskRun parentTaskRun,
+        Integer concurrency,
+        List<Dag.DagTask> taskDependencies
+    ) {
+        return resolveParallelNexts(
+            execution,
+            tasks, errors,
+            parentTaskRun,
+            concurrency,
+            (nextTaskRunStream, taskRuns) -> nextTaskRunStream
+                .filter(nextTaskRun -> {
+                    Task task = nextTaskRun.getTask();
+                    List<String> taskDependIds = taskDependencies
+                        .stream()
+                        .filter(taskDepend -> taskDepend
+                            .getTask()
+                            .getId()
+                            .equals(task.getId())
+                        )
+                        .findFirst()
+                        .get()
+                        .getDependsOn();
+
+                    // Check if have no dependencies OR all dependencies are terminated
+                    return taskDependIds == null ||
+                        new HashSet<>(taskRuns
+                            .stream()
+                            .filter(taskRun -> taskRun.getState().isTerminated())
+                            .map(TaskRun::getTaskId).toList()
+                        )
+                            .containsAll(taskDependIds);
+                })
+        );
+    }
+
+    public static List<NextTaskRun> resolveParallelNexts(
+        Execution execution,
+        List<ResolvedTask> tasks,
+        List<ResolvedTask> errors,
+        TaskRun parentTaskRun,
+        Integer concurrency,
+        BiFunction<Stream<NextTaskRun>, List<TaskRun>, Stream<NextTaskRun>> nextTaskRunFunction
     ) {
         if (execution.getState().getCurrent() == State.Type.KILLING) {
             return Collections.emptyList();
@@ -182,51 +240,51 @@ public static List<NextTaskRun> resolveParallelNexts(
                 .stream()
                 .map(resolvedTask -> resolvedTask.toNextTaskRun(execution));
 
+            nextTaskRunStream = nextTaskRunFunction.apply(nextTaskRunStream, taskRuns);
+
             if (concurrency > 0) {
                 nextTaskRunStream = nextTaskRunStream.limit(concurrency - runningCount);
             }
 
+
             return nextTaskRunStream.collect(Collectors.toList());
         }
 
         return Collections.emptyList();
     }
 
-    private final static TypeReference<List<Object>> TYPE_REFERENCE = new TypeReference<>() {};
+    private final static TypeReference<List<Object>> TYPE_REFERENCE = new TypeReference<>() {
+    };
     private final static ObjectMapper MAPPER = JacksonMapper.ofJson();
 
     @SuppressWarnings({"unchecked", "rawtypes"})
     public static List<ResolvedTask> resolveEachTasks(RunContext runContext, TaskRun parentTaskRun, List<Task> tasks, Object value) throws IllegalVariableEvaluationException {
         List<Object> values;
 
-        if(value instanceof String) {
+        if (value instanceof String) {
             String renderValue = runContext.render((String) value);
             try {
                 values = MAPPER.readValue(renderValue, TYPE_REFERENCE);
             } catch (JsonProcessingException e) {
                 throw new IllegalVariableEvaluationException(e);
             }
-        }
-        else if(value instanceof List) {
+        } else if (value instanceof List) {
             values = new ArrayList<>(((List<?>) value).size());
-            for(Object obj: (List<Object>) value) {
-                if(obj instanceof String){
+            for (Object obj : (List<Object>) value) {
+                if (obj instanceof String) {
                     values.add(runContext.render((String) obj));
                 }
                 else if(obj instanceof Map<?, ?>) {
                     //JSON or YAML map
                     values.add(runContext.render((Map) obj));
-                }
-                else {
+                } else {
                     throw new IllegalVariableEvaluationException("Unknown value element type: " + obj.getClass());
                 }
             }
-        }
-        else {
+        } else {
             throw new IllegalVariableEvaluationException("Unknown value type: " + value.getClass());
         }
 
-
         List<Object> distinctValue = values
             .stream()
             .distinct()
@@ -245,10 +303,10 @@ else if(obj instanceof Map<?, ?>) {
 
         ArrayList<ResolvedTask> result = new ArrayList<>();
 
-        for (Object current: distinctValue) {
-            for( Task task: tasks) {
+        for (Object current : distinctValue) {
+            for (Task task : tasks) {
                 try {
-                    String resolvedValue = current instanceof String ? (String)current : MAPPER.writeValueAsString(current);
+                    String resolvedValue = current instanceof String ? (String) current : MAPPER.writeValueAsString(current);
 
                     result.add(ResolvedTask.builder()
                         .task(task)
diff --git a/core/src/main/java/io/kestra/core/services/GraphService.java b/core/src/main/java/io/kestra/core/services/GraphService.java
index 7e6dc78f2a..25d302a093 100644
--- a/core/src/main/java/io/kestra/core/services/GraphService.java
+++ b/core/src/main/java/io/kestra/core/services/GraphService.java
@@ -8,6 +8,7 @@
 import io.kestra.core.models.tasks.FlowableTask;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.triggers.AbstractTrigger;
+import io.kestra.core.tasks.flows.Dag;
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.commons.lang3.tuple.Triple;
 
@@ -194,6 +195,21 @@ public static void ifElse(
         }
     }
 
+    public static void dag(
+        GraphCluster graph,
+        List<Dag.DagTask> tasks,
+        List<Task> errors,
+        TaskRun parent,
+        Execution execution
+    ) throws IllegalVariableEvaluationException {
+        fillGraphDag(graph, tasks, parent, execution);
+
+        // error cases
+        if (errors != null && errors.size() > 0) {
+            fillGraph(graph, errors, RelationType.ERROR, parent, execution, null);
+        }
+    }
+
     private static void iterate(
         GraphCluster graph,
         List<Task> tasks,
@@ -253,7 +269,7 @@ private static void fillGraph(
                     parentValues = execution.findChildsValues(currentTaskRun, true);
                 }
 
-                // detect kind
+                // detect kids
                 if (currentTask instanceof FlowableTask) {
                     FlowableTask<?> flowableTask = ((FlowableTask<?>) currentTask);
                     currentGraph = flowableTask.tasksTree(execution, currentTaskRun, parentValues);
@@ -316,6 +332,143 @@ private static void fillGraph(
         }
     }
 
+
+    private static void fillGraphDag(
+        GraphCluster graph,
+        List<Dag.DagTask> tasks,
+        TaskRun parent,
+        Execution execution
+    ) throws IllegalVariableEvaluationException {
+        List<GraphTask> nodeTaskCreated = new ArrayList<>();
+        List<String> nodeCreatedIds = new ArrayList<>();
+        Set<String> dependencies = tasks.stream().filter(taskDepend -> taskDepend.getDependsOn() != null).map(Dag.DagTask::getDependsOn).flatMap(Collection::stream).collect(Collectors.toSet());
+        AbstractGraph previous;
+
+        // we validate a GraphTask based on 3 nodes (root/ task / end)
+        if (graph.getGraph().nodes().size() >= 3 && new ArrayList<>(graph.getGraph().nodes()).get(2) instanceof GraphTask) {
+            previous = new ArrayList<>(graph.getGraph().nodes()).get(2);
+        } else {
+            previous = graph.getRoot();
+        }
+
+        while(nodeCreatedIds.size() < tasks.size()) {
+            Iterator<Dag.DagTask> iterator = tasks.stream().filter(taskDepend ->
+                // Check if the task has no dependencies OR all if its dependencies have been treated
+                (taskDepend.getDependsOn() == null || new HashSet<>(nodeCreatedIds).containsAll(taskDepend.getDependsOn()))
+                // AND if the task has not been treated yet
+                && !nodeCreatedIds.contains(taskDepend.getTask().getId())).iterator();
+            while (iterator.hasNext()) {
+                Dag.DagTask currentTask = iterator.next();
+                for (TaskRun currentTaskRun : findTaskRuns(currentTask.getTask(), execution, parent)) {
+                    AbstractGraph currentGraph;
+                    List<String> parentValues = null;
+
+                    RelationType newRelation = RelationType.PARALLEL;
+
+                    Relation relation = new Relation(
+                        newRelation,
+                        currentTaskRun == null ? null : currentTaskRun.getValue()
+                    );
+
+                    if (execution != null && currentTaskRun != null) {
+                        parentValues = execution.findChildsValues(currentTaskRun, true);
+                    }
+
+                    // detect kids
+                    if (currentTask.getTask() instanceof FlowableTask<?> flowableTask) {
+                        currentGraph = flowableTask.tasksTree(execution, currentTaskRun, parentValues);
+                    } else {
+                        currentGraph = new GraphTask(currentTask.getTask(), currentTaskRun, parentValues, RelationType.SEQUENTIAL);
+                    }
+
+                    // add the node
+                    graph.getGraph().addNode(currentGraph);
+
+                    // link to previous one
+                    if(currentTask.getDependsOn() == null) {
+                        graph.getGraph().addEdge(
+                            previous,
+                            currentGraph instanceof GraphCluster ? ((GraphCluster) currentGraph).getRoot() : currentGraph,
+                            relation
+                        );
+                    } else {
+                        for (String dependsOn : currentTask.getDependsOn()) {
+                            GraphTask previousNode = nodeTaskCreated.stream().filter(node -> node.getTask().getId().equals(dependsOn)).findFirst().orElse(null);
+                            if(previousNode!= null && !previousNode.getTask().isFlowable()) {
+                                graph.getGraph().addEdge(
+                                    previousNode,
+                                    currentGraph instanceof GraphCluster ? ((GraphCluster) currentGraph).getRoot() : currentGraph,
+                                    relation
+                                );
+                            } else {
+                                graph.getGraph()
+                                    .nodes()
+                                    .stream()
+                                    .filter(node -> node.getUid().equals("cluster_" + dependsOn))
+                                    .findFirst()
+                                    .ifPresent(previousClusterNodeEnd -> graph.getGraph().addEdge(
+                                        ((GraphCluster) previousClusterNodeEnd).getEnd(),
+                                        currentGraph instanceof GraphCluster ? ((GraphCluster) currentGraph).getRoot() : currentGraph,
+                                        relation
+                                ));
+                            }
+                        }
+                    }
+                    // link to last one if task isn't a dependency
+                    if (!dependencies.contains(currentTask.getTask().getId())) {
+                        if(currentTask.getTask() instanceof FlowableTask<?> flowableTask) {
+                            graph.getGraph().addEdge(
+                                ((GraphCluster) currentGraph).getEnd(),
+                                graph.getEnd(),
+                                new Relation()
+                            );
+                        } else {
+                            graph.getGraph().addEdge(
+                                currentGraph,
+                                graph.getEnd(),
+                                new Relation()
+                            );
+                        }
+                    }
+
+                    if(currentGraph instanceof GraphTask) {
+                        nodeTaskCreated.add((GraphTask) currentGraph);
+                    }
+                    nodeCreatedIds.add(currentTask.getTask().getId());
+                }
+            }
+        }
+    }
+
+
+    private static void generatePath(Dag.DagTask task, Map<String, Dag.DagTask> taskMap, List<Dag.DagTask> currentPath,
+                                     List<List<Task>> allPaths, Set<String> visited) {
+        currentPath.add(task);
+        visited.add(task.getTask().getId());
+
+        if (taskMap.containsKey(task.getTask().getId())) {
+            List<String> dependsOnIds = task.getDependsOn();
+
+            if(dependsOnIds != null) {
+                for (String dependencyId : dependsOnIds) {
+                    Dag.DagTask dependencyTask = taskMap.get(dependencyId);
+                    if (!visited.contains(dependencyId)) {
+                        generatePath(dependencyTask, taskMap, currentPath, allPaths, visited);
+                    }
+                }
+            }
+        }
+
+        if (task.getDependsOn() == null) {
+            Collections.reverse(currentPath);
+            allPaths.add(currentPath.stream().map(Dag.DagTask::getTask).collect(Collectors.toList()));
+        }
+
+        currentPath.remove(currentPath.size() - 1);
+        visited.remove(task.getTask().getId());
+    }
+
+
     private static boolean isAllLinkToEnd(RelationType relationType) {
         return relationType == RelationType.PARALLEL || relationType == RelationType.CHOICE;
     }
diff --git a/core/src/main/java/io/kestra/core/tasks/flows/Dag.java b/core/src/main/java/io/kestra/core/tasks/flows/Dag.java
new file mode 100644
index 0000000000..cd6fffd892
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/tasks/flows/Dag.java
@@ -0,0 +1,181 @@
+package io.kestra.core.tasks.flows;
+
+import io.kestra.core.exceptions.IllegalVariableEvaluationException;
+import io.kestra.core.models.annotations.PluginProperty;
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.executions.NextTaskRun;
+import io.kestra.core.models.executions.TaskRun;
+import io.kestra.core.models.hierarchies.GraphCluster;
+import io.kestra.core.models.hierarchies.RelationType;
+import io.kestra.core.models.tasks.*;
+import io.kestra.core.runners.FlowableUtils;
+import io.kestra.core.runners.RunContext;
+import io.kestra.core.services.GraphService;
+import io.kestra.core.validations.DagTaskValidation;
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.*;
+import lombok.experimental.SuperBuilder;
+
+import javax.validation.Valid;
+import javax.validation.constraints.NotBlank;
+import javax.validation.constraints.NotEmpty;
+import javax.validation.constraints.NotNull;
+import java.util.*;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+@SuperBuilder
+@ToString
+@EqualsAndHashCode
+@Getter
+@NoArgsConstructor
+@DagTaskValidation
+@Schema(
+    title = "Create a directed acyclic graph (DAG) flow without bothering with the graph structure.",
+    description = "List your tasks and their dependencies, and Kestra will figure out the rest.\n" +
+        "Task can only depends on task from the DAG tasks.\n" +
+        "For technical reasons, low-code interaction with this Task is disabled for now."
+)
+public class Dag extends Task implements FlowableTask<VoidOutput> {
+    @NotNull
+    @NotBlank
+    @Builder.Default
+    @Schema(
+        title = "Number of concurrent parallel tasks",
+        description = "If the value is `0`, no limit exist and all the tasks will start at the same time"
+    )
+    @PluginProperty
+    private final Integer concurrent = 0;
+
+    @NotEmpty
+    private List<DagTask> tasks;
+
+    @Valid
+    @PluginProperty
+    protected List<Task> errors;
+
+    @Override
+    public GraphCluster tasksTree(Execution execution, TaskRun taskRun, List<String> parentValues) throws IllegalVariableEvaluationException {
+        GraphCluster subGraph = new GraphCluster(this, taskRun, parentValues, RelationType.DYNAMIC);
+
+        this.controlTask();
+
+        GraphService.dag(
+            subGraph,
+            this.getTasks(),
+            this.errors,
+            taskRun,
+            execution
+        );
+
+        return subGraph;
+    }
+
+    private void controlTask() throws IllegalVariableEvaluationException {
+        List<String> dagCheckNotExistTasks = this.dagCheckNotExistTask(this.tasks);
+        if (!dagCheckNotExistTasks.isEmpty()) {
+            throw new IllegalVariableEvaluationException("Some task doesn't exists on task '" + this.id + "': " +  String.join(", ", dagCheckNotExistTasks));
+        }
+
+        ArrayList<String> cyclicDependenciesTasks = this.dagCheckCyclicDependencies(this.tasks);
+        if (!cyclicDependenciesTasks.isEmpty()) {
+            throw new IllegalVariableEvaluationException("Infinite loop detected on task '" + this.id + "': " + String.join(", ", cyclicDependenciesTasks));
+        }
+    }
+
+    @Override
+    public List<Task> allChildTasks() {
+        return Stream
+            .concat(
+                this.tasks != null ? this.tasks.stream().map(DagTask::getTask) : Stream.empty(),
+                this.errors != null ? this.errors.stream() : Stream.empty()
+            )
+            .collect(Collectors.toList());
+    }
+
+    @Override
+    public List<ResolvedTask> childTasks(RunContext runContext, TaskRun parentTaskRun) throws IllegalVariableEvaluationException {
+        return FlowableUtils.resolveTasks(this.tasks.stream().map(DagTask::getTask).toList(), parentTaskRun);
+    }
+
+    @Override
+    public List<NextTaskRun> resolveNexts(RunContext runContext, Execution execution, TaskRun parentTaskRun) throws IllegalVariableEvaluationException {
+        this.controlTask();
+
+        return FlowableUtils.resolveDagNexts(
+            execution,
+            this.childTasks(runContext, parentTaskRun),
+            FlowableUtils.resolveTasks(this.errors, parentTaskRun),
+            parentTaskRun,
+            this.concurrent,
+            this.tasks
+        );
+    }
+
+    public List<String> dagCheckNotExistTask(List<DagTask> taskDepends) {
+        List<String> dependenciesIds = taskDepends
+            .stream()
+            .map(DagTask::getDependsOn)
+            .filter(Objects::nonNull)
+            .flatMap(Collection::stream)
+            .toList();
+
+        List<String> tasksIds = taskDepends
+            .stream()
+            .map(taskDepend -> taskDepend.getTask().getId())
+            .toList();
+
+        return dependenciesIds.stream()
+            .filter(dependencyId -> !tasksIds.contains(dependencyId))
+            .collect(Collectors.toList());
+    }
+
+    public ArrayList<String> dagCheckCyclicDependencies(List<DagTask> taskDepends) {
+        ArrayList<String> cyclicDependency = new ArrayList<>();
+        taskDepends.forEach(taskDepend -> {
+            if (taskDepend.getDependsOn() != null) {
+                List<String> nestedDependencies = this.nestedDependencies(taskDepend, taskDepends, new ArrayList<>());
+                if (nestedDependencies.contains(taskDepend.getTask().getId())) {
+                    cyclicDependency.add(taskDepend.getTask().getId());
+                }
+            }
+        });
+
+        return cyclicDependency;
+    }
+
+    private ArrayList<String> nestedDependencies(DagTask taskDepend, List<DagTask> tasks, List<String> visited) {
+        final ArrayList<String> localVisited = new ArrayList<>(visited);
+        if (taskDepend.getDependsOn() != null) {
+            taskDepend.getDependsOn()
+                .stream()
+                .filter(depend -> !localVisited.contains(depend))
+                .forEach(depend -> {
+                    localVisited.add(depend);
+                    Optional<DagTask> task = tasks
+                        .stream()
+                        .filter(t -> t.getTask().getId().equals(depend))
+                        .findFirst();
+
+                    if (task.isPresent()) {
+                        localVisited.addAll(this.nestedDependencies(task.get(), tasks, localVisited));
+                    }
+                });
+        }
+        return localVisited;
+    }
+
+    @SuperBuilder
+    @ToString
+    @EqualsAndHashCode
+    @Getter
+    @NoArgsConstructor
+    public static class DagTask {
+        @NotNull
+        @PluginProperty
+        private Task task;
+
+        @PluginProperty
+        private List<String> dependsOn;
+    }
+}
diff --git a/core/src/main/java/io/kestra/core/validations/DagTaskValidation.java b/core/src/main/java/io/kestra/core/validations/DagTaskValidation.java
new file mode 100644
index 0000000000..6a391c6c4b
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/validations/DagTaskValidation.java
@@ -0,0 +1,11 @@
+package io.kestra.core.validations;
+
+import javax.validation.Constraint;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Constraint(validatedBy = { })
+public @interface DagTaskValidation {
+    String message() default "invalid Dag task";
+}
diff --git a/core/src/main/java/io/kestra/core/validations/ValidationFactory.java b/core/src/main/java/io/kestra/core/validations/ValidationFactory.java
index 32faf52460..54218041a1 100644
--- a/core/src/main/java/io/kestra/core/validations/ValidationFactory.java
+++ b/core/src/main/java/io/kestra/core/validations/ValidationFactory.java
@@ -6,6 +6,7 @@
 import io.kestra.core.models.flows.Input;
 import io.kestra.core.models.tasks.RunnableTask;
 import io.kestra.core.models.tasks.Task;
+import io.kestra.core.tasks.flows.Dag;
 import io.kestra.core.models.tasks.WorkerGroup;
 import io.kestra.core.tasks.flows.Switch;
 import io.kestra.core.tasks.flows.WorkingDirectory;
@@ -14,6 +15,7 @@
 import jakarta.inject.Named;
 import jakarta.inject.Singleton;
 
+import javax.validation.ConstraintViolation;
 import java.io.IOException;
 import java.text.SimpleDateFormat;
 import java.util.*;
@@ -116,6 +118,42 @@ ConstraintValidator<SwitchTaskValidation, Switch> switchTaskValidation() {
         };
     }
 
+    @Singleton
+    ConstraintValidator<DagTaskValidation, Dag> dagTaskValidation() {
+        return (value, annotationMetadata, context) -> {
+            if (value == null) {
+                return true;
+            }
+
+            if (value.getTasks() == null || value.getTasks().isEmpty()) {
+                context.messageTemplate("No task defined");
+
+                return false;
+            }
+
+            List<Dag.DagTask> taskDepends = value.getTasks();
+
+            // Check for not existing taskId
+            List<String> invalidDependencyIds = value.dagCheckNotExistTask(taskDepends);
+            if (!invalidDependencyIds.isEmpty()) {
+                String errorMessage = "Not existing task id in dependency: " + String.join(", ", invalidDependencyIds);
+                context.messageTemplate(errorMessage);
+
+                return false;
+            }
+
+            // Check for cyclic dependencies
+            ArrayList<String> cyclicDependency = value.dagCheckCyclicDependencies(taskDepends);
+            if (!cyclicDependency.isEmpty()) {
+                context.messageTemplate("Cyclic dependency detected: " + String.join(", ", cyclicDependency));
+
+                return false;
+            }
+
+            return true;
+        };
+    }
+
     @Singleton
     ConstraintValidator<WorkingDirectoryTaskValidation, WorkingDirectory> workingDirectoryTaskValidation() {
         return (value, annotationMetadata, context) -> {
diff --git a/ui/src/components/flows/tasks/Task.js b/ui/src/components/flows/tasks/Task.js
index 0253555156..81cf131868 100644
--- a/ui/src/components/flows/tasks/Task.js
+++ b/ui/src/components/flows/tasks/Task.js
@@ -39,7 +39,7 @@ export default {
             }
 
             if (Object.prototype.hasOwnProperty.call(property, "$ref")) {
-                if (property.$ref.includes("Task")) {
+                if (property.$ref.includes("tasks.Task")) {
                     return "task"
                 }
 
diff --git a/ui/src/components/graph/nodes/Edge.vue b/ui/src/components/graph/nodes/Edge.vue
index 8be14a7057..b52e704d74 100644
--- a/ui/src/components/graph/nodes/Edge.vue
+++ b/ui/src/components/graph/nodes/Edge.vue
@@ -209,7 +209,7 @@
 
     <EdgeLabelRenderer style="z-index: 10">
         <div
-            v-if="getEdgeLabel(props.data.edge.relation) !== ''"
+            v-if="getEdgeLabel(props.data.edge.relation) !== '' && !props.data.disabled"
             @mouseover="onMouseOver"
             @mouseleave="onMouseLeave"
             :style="{
diff --git a/ui/src/components/inputs/LowCodeEditor.vue b/ui/src/components/inputs/LowCodeEditor.vue
index 7af6157efe..12df1a0122 100644
--- a/ui/src/components/inputs/LowCodeEditor.vue
+++ b/ui/src/components/inputs/LowCodeEditor.vue
@@ -449,6 +449,8 @@
                 })
             }
 
+            let disabledLowCode = [];
+
             for (const node of props.flowGraph.nodes) {
                 const dagreNode = dagreGraph.node(node.uid);
                 let nodeType = "task";
@@ -461,6 +463,15 @@
                 } else if (node.type.includes("GraphTrigger")) {
                     nodeType = "trigger";
                 }
+                // Disable interaction for Dag task
+                // because our low code editor can not handle it for now
+                if (isTaskNode(node) && node.task.type === "io.kestra.core.tasks.flows.Dag") {
+                    disabledLowCode.push(node.task.id);
+                    YamlUtils.getChildrenTasks(props.source, node.task.id).forEach(child => {
+                        disabledLowCode.push(child);
+                    })
+                }
+
                 elements.value.push({
                     id: node.uid,
                     label: isTaskNode(node) ? node.task.id : "",
@@ -473,7 +484,7 @@
                     sourcePosition: isHorizontal.value ? Position.Right : Position.Bottom,
                     targetPosition: isHorizontal.value ? Position.Left : Position.Top,
                     parentNode: clusters[node.uid] ? clusters[node.uid].uid : undefined,
-                    draggable: nodeType === "task" && !props.isReadOnly,
+                    draggable: nodeType === "task" && !props.isReadOnly && isTaskNode(node) ? !disabledLowCode.includes(node.task.id) : false,
                     data: {
                         node: node,
                         namespace: props.namespace,
@@ -495,7 +506,8 @@
                         edge: edge,
                         haveAdd: complexEdgeHaveAdd(edge),
                         isFlowable: flowables().includes(edge.source) || flowables().includes(edge.target),
-                        nextTaskId: getNextTaskId(edge.target)
+                        nextTaskId: getNextTaskId(edge.target),
+                        disabled: disabledLowCode.includes(edge.source)
                     }
                 })
             }
diff --git a/ui/src/utils/yamlUtils.js b/ui/src/utils/yamlUtils.js
index 7c6b484cdc..cfc91337d7 100644
--- a/ui/src/utils/yamlUtils.js
+++ b/ui/src/utils/yamlUtils.js
@@ -403,6 +403,19 @@ export default class YamlUtils {
         return isChildrenOf;
     }
 
+    static getChildrenTasks(source, taskId) {
+        const yamlDoc = yaml.parseDocument(YamlUtils.extractTask(source, taskId));
+        const children = [];
+        yaml.visit(yamlDoc, {
+            Map(_, map) {
+                if (map.get("id") !== taskId) {
+                    children.push(map.get("id"));
+                }
+            }
+        })
+        return children;
+    }
+
     static replaceIdAndNamespace(source, id, namespace) {
         return source.replace(/^(id\s*:\s*(["']?))\S*/m, "$1"+id+"$2").replace(/^(namespace\s*:\s*(["']?))\S*/m, "$1"+namespace+"$2")
     }
diff --git a/webserver/src/main/java/io/kestra/webserver/controllers/FlowController.java b/webserver/src/main/java/io/kestra/webserver/controllers/FlowController.java
index af3e7aac0b..2118132719 100644
--- a/webserver/src/main/java/io/kestra/webserver/controllers/FlowController.java
+++ b/webserver/src/main/java/io/kestra/webserver/controllers/FlowController.java
@@ -472,7 +472,7 @@ public List<ValidateConstraintViolation> validateFlows(
     @Operation(tags = {"Flows"}, summary = "Validate a list of flows")
     public ValidateConstraintViolation validateTask(
         @Parameter(description = "A list of flows") @Body String task,
-        @Parameter(description = "Type of task") TaskValidationType section
+        @Parameter(description = "Type of task") @QueryValue TaskValidationType section
     ) {
         ValidateConstraintViolation.ValidateConstraintViolationBuilder<?, ?> validateConstraintViolationBuilder = ValidateConstraintViolation.builder();
 


Test patch:
diff --git a/core/src/test/java/io/kestra/core/tasks/flows/DagTest.java b/core/src/test/java/io/kestra/core/tasks/flows/DagTest.java
new file mode 100644
index 0000000000..44f352395c
--- /dev/null
+++ b/core/src/test/java/io/kestra/core/tasks/flows/DagTest.java
@@ -0,0 +1,75 @@
+package io.kestra.core.tasks.flows;
+
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.flows.State;
+import io.kestra.core.models.validations.ModelValidator;
+import io.kestra.core.runners.AbstractMemoryRunnerTest;
+import io.kestra.core.serializers.YamlFlowParser;
+import io.kestra.core.utils.TestsUtils;
+import jakarta.inject.Inject;
+import org.junit.jupiter.api.Test;
+
+import javax.validation.ConstraintViolationException;
+import java.io.File;
+import java.net.URL;
+import java.util.Optional;
+import java.util.concurrent.TimeoutException;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.is;
+
+public class DagTest extends AbstractMemoryRunnerTest {
+    @Inject
+    YamlFlowParser yamlFlowParser = new YamlFlowParser();
+
+    @Inject
+    ModelValidator modelValidator;
+
+    @Test
+    void dag() throws TimeoutException {
+        Execution execution = runnerUtils.runOne(
+            "io.kestra.tests",
+            "dag",
+            null,
+            null
+        );
+
+        assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
+        assertThat(execution.getTaskRunList().size(), is(7));
+        assertThat(execution.getTaskRunList().get(1).getTaskId(), is("task1"));
+        assertThat(execution.getTaskRunList().get(6).getTaskId(), is("task5"));
+    }
+
+    @Test
+    void dagCyclicDependencies() throws TimeoutException {
+        Flow flow = this.parse("flows/invalids/dag-cyclicdependency.yaml");
+        Optional<ConstraintViolationException> validate = modelValidator.isValid(flow);
+
+        assertThat(validate.isPresent(), is(true));
+        assertThat(validate.get().getConstraintViolations().size(), is(1));
+
+        assertThat(validate.get().getMessage(), containsString("tasks[0]: Cyclic dependency detected: task1, task2"));
+    }
+
+    @Test
+    void dagNotExistTask() throws TimeoutException {
+        Flow flow = this.parse("flows/invalids/dag-notexist-task.yaml");
+        Optional<ConstraintViolationException> validate = modelValidator.isValid(flow);
+
+        assertThat(validate.isPresent(), is(true));
+        assertThat(validate.get().getConstraintViolations().size(), is(1));
+
+        assertThat(validate.get().getMessage(), containsString("tasks[0]: Not existing task id in dependency: taskX"));
+    }
+
+    private Flow parse(String path) {
+        URL resource = TestsUtils.class.getClassLoader().getResource(path);
+        assert resource != null;
+
+        File file = new File(resource.getFile());
+
+        return yamlFlowParser.parse(file, Flow.class);
+    }
+}
diff --git a/core/src/test/resources/flows/invalids/dag-cyclicdependency.yaml b/core/src/test/resources/flows/invalids/dag-cyclicdependency.yaml
new file mode 100644
index 0000000000..54cc832d00
--- /dev/null
+++ b/core/src/test/resources/flows/invalids/dag-cyclicdependency.yaml
@@ -0,0 +1,38 @@
+id: "dag-cyclicdependency"
+namespace: io.kestra.tests
+tasks:
+  - id: dag
+    description: "my task"
+    type: io.kestra.core.tasks.flows.Dag
+    tasks:
+      - task:
+          id: task1
+          type: io.kestra.core.tasks.log.Log
+          message: "1 !"
+        dependsOn:
+          - task2
+      - task:
+          id: task2
+          type: io.kestra.core.tasks.log.Log
+          message: "2 !"
+        dependsOn:
+          - task1
+      - task:
+          id: task3
+          type: io.kestra.core.tasks.log.Log
+          message: "1 !"
+        dependsOn:
+          - task2
+      - task:
+          id: task4
+          type: io.kestra.core.tasks.log.Log
+          message: "1 !"
+        dependsOn:
+          - task2
+      - task:
+          id: task5
+          type: io.kestra.core.tasks.log.Log
+          message: "1 !"
+        dependsOn:
+          - task4
+          - task3
diff --git a/core/src/test/resources/flows/invalids/dag-notexist-task.yaml b/core/src/test/resources/flows/invalids/dag-notexist-task.yaml
new file mode 100644
index 0000000000..b45f734c84
--- /dev/null
+++ b/core/src/test/resources/flows/invalids/dag-notexist-task.yaml
@@ -0,0 +1,36 @@
+id: "dag-notexist-task"
+namespace: io.kestra.tests
+tasks:
+  - id: dag
+    description: "my task"
+    type: io.kestra.core.tasks.flows.Dag
+    tasks:
+      - task:
+          id: task1
+          type: io.kestra.core.tasks.log.Log
+          message: "1 !"
+      - task:
+          id: task2
+          type: io.kestra.core.tasks.log.Log
+          message: "2 !"
+        dependsOn:
+          - taskX
+      - task:
+          id: task3
+          type: io.kestra.core.tasks.log.Log
+          message: "1 !"
+        dependsOn:
+          - task2
+      - task:
+          id: task4
+          type: io.kestra.core.tasks.log.Log
+          message: "1 !"
+        dependsOn:
+          - task2
+      - task:
+          id: task5
+          type: io.kestra.core.tasks.log.Log
+          message: "1 !"
+        dependsOn:
+          - task4
+          - task3
diff --git a/core/src/test/resources/flows/valids/dag.yaml b/core/src/test/resources/flows/valids/dag.yaml
new file mode 100644
index 0000000000..6ddb6f732f
--- /dev/null
+++ b/core/src/test/resources/flows/valids/dag.yaml
@@ -0,0 +1,51 @@
+id: "dag"
+namespace: "io.kestra.tests"
+tasks:
+  - id: dag
+    description: "my task"
+    type: io.kestra.core.tasks.flows.Dag
+    tasks:
+      - task:
+          id: task1
+          type: io.kestra.core.tasks.scripts.Bash
+          commands:
+            - sleep 1
+      - task:
+          id: task2
+          type: io.kestra.core.tasks.scripts.Bash
+          commands:
+            - sleep 1
+            - echo '::{"outputs":{"test":"value","int":2,"bool":true,"float":3.65}}::'
+        dependsOn:
+          - task1
+      - task:
+          id: task3
+          type: io.kestra.core.tasks.scripts.Bash
+          commands:
+            - sleep 1
+        dependsOn:
+          - task1
+      - task:
+          id: task4
+          type: io.kestra.core.tasks.scripts.Bash
+          commands:
+            - sleep 1
+        dependsOn:
+          - task2
+      - task:
+          id: task5
+          type: io.kestra.core.tasks.scripts.Bash
+          commands:
+            - sleep 1
+        dependsOn:
+          - task4
+          - task3
+      - task:
+          id: task6
+          type: io.kestra.core.tasks.scripts.Bash
+          commands:
+            - sleep 1
+            - echo {{ outputs.task2.vars.test }}
+        dependsOn:
+          - task2
+          - task3