Repository: serverlessworkflow/sdk-java
Pull request #483 (2024-11-26 16:40:11)

[Fix_#474] For task implementation
Fix https://github.com/serverlessworkflow/sdk-java/issues/474


Patch:
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/ContextAware.java b/impl/core/src/main/java/io/serverlessworkflow/impl/ContextAware.java
new file mode 100644
index 0000000..a58dc34
--- /dev/null
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/ContextAware.java
@@ -0,0 +1,22 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.impl;
+
+import java.util.Map;
+
+public interface ContextAware {
+  Map<String, Object> variables();
+}
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/DefaultWorkflowPosition.java b/impl/core/src/main/java/io/serverlessworkflow/impl/DefaultWorkflowPosition.java
index 2e51f6a..54f993b 100644
--- a/impl/core/src/main/java/io/serverlessworkflow/impl/DefaultWorkflowPosition.java
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/DefaultWorkflowPosition.java
@@ -17,7 +17,19 @@ package io.serverlessworkflow.impl;
 
 public class DefaultWorkflowPosition implements WorkflowPosition {
 
-  private StringBuilder sb = new StringBuilder("");
+  private StringBuilder sb;
+
+  DefaultWorkflowPosition() {
+    this.sb = new StringBuilder("");
+  }
+
+  private DefaultWorkflowPosition(WorkflowPosition position) {
+    this.sb = new StringBuilder(position.toString());
+  }
+
+  public DefaultWorkflowPosition copy() {
+    return new DefaultWorkflowPosition(this);
+  }
 
   @Override
   public WorkflowPosition addIndex(int index) {
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/DefaultWorkflowPositionFactory.java b/impl/core/src/main/java/io/serverlessworkflow/impl/DefaultWorkflowPositionFactory.java
new file mode 100644
index 0000000..00b0085
--- /dev/null
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/DefaultWorkflowPositionFactory.java
@@ -0,0 +1,32 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.impl;
+
+class DefaultWorkflowPositionFactory implements WorkflowPositionFactory {
+
+  private static WorkflowPositionFactory instance = new DefaultWorkflowPositionFactory();
+
+  public static WorkflowPositionFactory get() {
+    return instance;
+  }
+
+  private DefaultWorkflowPositionFactory() {}
+
+  @Override
+  public WorkflowPosition buildPosition() {
+    return new DefaultWorkflowPosition();
+  }
+}
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/TaskContext.java b/impl/core/src/main/java/io/serverlessworkflow/impl/TaskContext.java
index 91f6aa6..c23de49 100644
--- a/impl/core/src/main/java/io/serverlessworkflow/impl/TaskContext.java
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/TaskContext.java
@@ -19,28 +19,48 @@ import com.fasterxml.jackson.databind.JsonNode;
 import io.serverlessworkflow.api.types.FlowDirective;
 import io.serverlessworkflow.api.types.FlowDirectiveEnum;
 import io.serverlessworkflow.api.types.TaskBase;
+import java.util.HashMap;
+import java.util.Map;
 
-public class TaskContext<T extends TaskBase> {
+public class TaskContext<T extends TaskBase> implements ContextAware {
 
   private final JsonNode rawInput;
   private final T task;
+  private final WorkflowPosition position;
 
   private JsonNode input;
   private JsonNode output;
   private JsonNode rawOutput;
   private FlowDirective flowDirective;
+  private Map<String, Object> contextVariables;
 
-  public TaskContext(JsonNode rawInput, T task) {
-    this.rawInput = rawInput;
+  public TaskContext(JsonNode input, WorkflowPosition position) {
+    this.rawInput = input;
+    this.position = position;
+    this.task = null;
+    this.contextVariables = new HashMap<>();
+    init();
+  }
+
+  public TaskContext(JsonNode input, TaskContext<?> taskContext, T task) {
+    this.rawInput = input;
+    this.position = taskContext.position.copy();
+    this.task = task;
+    this.flowDirective = task.getThen();
+    this.contextVariables = new HashMap<>(taskContext.variables());
+    init();
+  }
+
+  private void init() {
     this.input = rawInput;
     this.rawOutput = rawInput;
     this.output = rawInput;
-    this.task = task;
-    this.flowDirective = task.getThen();
   }
 
   public void input(JsonNode input) {
     this.input = input;
+    this.rawOutput = input;
+    this.output = input;
   }
 
   public JsonNode input() {
@@ -81,4 +101,12 @@ public class TaskContext<T extends TaskBase> {
         ? new FlowDirective().withFlowDirectiveEnum(FlowDirectiveEnum.CONTINUE)
         : flowDirective;
   }
+
+  public Map<String, Object> variables() {
+    return contextVariables;
+  }
+
+  public WorkflowPosition position() {
+    return position;
+  }
 }
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowApplication.java b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowApplication.java
index 02d807f..d9da16b 100644
--- a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowApplication.java
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowApplication.java
@@ -25,7 +25,6 @@ import io.serverlessworkflow.impl.jsonschema.DefaultSchemaValidatorFactory;
 import io.serverlessworkflow.impl.jsonschema.SchemaValidatorFactory;
 import io.serverlessworkflow.impl.resources.DefaultResourceLoaderFactory;
 import io.serverlessworkflow.impl.resources.ResourceLoaderFactory;
-
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
@@ -40,17 +39,20 @@ public class WorkflowApplication implements AutoCloseable {
   private final SchemaValidatorFactory schemaValidatorFactory;
   private final Collection<WorkflowExecutionListener> listeners;
   private final Map<WorkflowId, WorkflowDefinition> definitions;
+  private final WorkflowPositionFactory positionFactory;
 
   public WorkflowApplication(
       TaskExecutorFactory taskFactory,
       ExpressionFactory exprFactory,
       ResourceLoaderFactory resourceLoaderFactory,
       SchemaValidatorFactory schemaValidatorFactory,
+      WorkflowPositionFactory positionFactory,
       Collection<WorkflowExecutionListener> listeners) {
     this.taskFactory = taskFactory;
     this.exprFactory = exprFactory;
     this.resourceLoaderFactory = resourceLoaderFactory;
     this.schemaValidatorFactory = schemaValidatorFactory;
+    this.positionFactory = positionFactory;
     this.listeners = listeners;
     this.definitions = new ConcurrentHashMap<>();
   }
@@ -85,6 +87,7 @@ public class WorkflowApplication implements AutoCloseable {
     private Collection<WorkflowExecutionListener> listeners;
     private ResourceLoaderFactory resourceLoaderFactory = DefaultResourceLoaderFactory.get();
     private SchemaValidatorFactory schemaValidatorFactory = DefaultSchemaValidatorFactory.get();
+    private WorkflowPositionFactory positionFactory = DefaultWorkflowPositionFactory.get();
 
     private Builder() {}
 
@@ -111,6 +114,11 @@ public class WorkflowApplication implements AutoCloseable {
       return this;
     }
 
+    public Builder withPositionFactory(WorkflowPositionFactory positionFactory) {
+      this.positionFactory = positionFactory;
+      return this;
+    }
+
     public Builder withSchemaValidatorFactory(SchemaValidatorFactory factory) {
       this.schemaValidatorFactory = factory;
       return this;
@@ -122,6 +130,7 @@ public class WorkflowApplication implements AutoCloseable {
           exprFactory,
           resourceLoaderFactory,
           schemaValidatorFactory,
+          positionFactory,
           listeners == null
               ? Collections.emptySet()
               : Collections.unmodifiableCollection(listeners));
@@ -146,4 +155,8 @@ public class WorkflowApplication implements AutoCloseable {
     }
     definitions.clear();
   }
+
+  public WorkflowPositionFactory positionFactory() {
+    return positionFactory;
+  }
 }
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowContext.java b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowContext.java
index 4f0f0f1..d5c1f42 100644
--- a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowContext.java
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowContext.java
@@ -19,49 +19,16 @@ import com.fasterxml.jackson.databind.JsonNode;
 import io.serverlessworkflow.impl.json.JsonUtils;
 
 public class WorkflowContext {
-  private final WorkflowPosition position;
   private final WorkflowDefinition definition;
   private final JsonNode input;
-  private JsonNode current;
   private JsonNode context;
 
-  private WorkflowContext(
-      WorkflowPosition position, WorkflowDefinition definition, JsonNode input) {
-    this.position = position;
+  WorkflowContext(WorkflowDefinition definition, JsonNode input) {
     this.definition = definition;
     this.input = input;
-    this.current = input.deepCopy();
     this.context = JsonUtils.mapper().createObjectNode();
   }
 
-  public static Builder builder(WorkflowDefinition definition, JsonNode input) {
-    return new Builder(definition, input);
-  }
-
-  public static class Builder {
-    private WorkflowPosition position = new DefaultWorkflowPosition();
-    private WorkflowDefinition definition;
-    private JsonNode input;
-
-    private Builder(WorkflowDefinition definition, JsonNode input) {
-      this.definition = definition;
-      this.input = input;
-    }
-
-    public Builder position(WorkflowPosition position) {
-      this.position = position;
-      return this;
-    }
-
-    public WorkflowContext build() {
-      return new WorkflowContext(position, definition, input);
-    }
-  }
-
-  public WorkflowPosition position() {
-    return position;
-  }
-
   public JsonNode context() {
     return context;
   }
@@ -74,14 +41,6 @@ public class WorkflowContext {
     return input;
   }
 
-  public void current(JsonNode output) {
-    this.current = output;
-  }
-
-  public JsonNode current() {
-    return current;
-  }
-
   public WorkflowDefinition definition() {
     return definition;
   }
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowDefinition.java b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowDefinition.java
index f593cea..3a76ff1 100644
--- a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowDefinition.java
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowDefinition.java
@@ -28,7 +28,6 @@ import io.serverlessworkflow.impl.json.JsonUtils;
 import io.serverlessworkflow.impl.jsonschema.SchemaValidator;
 import io.serverlessworkflow.impl.jsonschema.SchemaValidatorFactory;
 import io.serverlessworkflow.impl.resources.ResourceLoader;
-
 import java.nio.file.Path;
 import java.util.Collection;
 import java.util.Map;
@@ -47,6 +46,7 @@ public class WorkflowDefinition implements AutoCloseable {
   private final ExpressionFactory exprFactory;
   private final ResourceLoader resourceLoader;
   private final SchemaValidatorFactory schemaValidatorFactory;
+  private final WorkflowPositionFactory positionFactory;
   private final Map<String, TaskExecutor<? extends TaskBase>> taskExecutors =
       new ConcurrentHashMap<>();
 
@@ -56,12 +56,14 @@ public class WorkflowDefinition implements AutoCloseable {
       TaskExecutorFactory taskFactory,
       ResourceLoader resourceLoader,
       ExpressionFactory exprFactory,
-      SchemaValidatorFactory schemaValidatorFactory) {
+      SchemaValidatorFactory schemaValidatorFactory,
+      WorkflowPositionFactory positionFactory) {
     this.workflow = workflow;
     this.listeners = listeners;
     this.taskFactory = taskFactory;
     this.exprFactory = exprFactory;
     this.schemaValidatorFactory = schemaValidatorFactory;
+    this.positionFactory = positionFactory;
     this.resourceLoader = resourceLoader;
     if (workflow.getInput() != null) {
       Input input = workflow.getInput();
@@ -90,7 +92,8 @@ public class WorkflowDefinition implements AutoCloseable {
         application.taskFactory(),
         application.resourceLoaderFactory().getResourceLoader(path),
         application.expressionFactory(),
-        application.validatorFactory());
+        application.validatorFactory(),
+        application.positionFactory());
   }
 
   public WorkflowInstance execute(Object input) {
@@ -142,6 +145,10 @@ public class WorkflowDefinition implements AutoCloseable {
     return resourceLoader;
   }
 
+  public WorkflowPositionFactory positionFactory() {
+    return positionFactory;
+  }
+
   @Override
   public void close() {
     // TODO close resourcers hold for uncompleted process instances, if any
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowFilter.java b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowFilter.java
index 7fde97b..7d25df4 100644
--- a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowFilter.java
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowFilter.java
@@ -16,9 +16,8 @@
 package io.serverlessworkflow.impl;
 
 import com.fasterxml.jackson.databind.JsonNode;
-import java.util.Optional;
 
 @FunctionalInterface
 public interface WorkflowFilter {
-  JsonNode apply(WorkflowContext workflow, Optional<TaskContext<?>> task, JsonNode node);
+  JsonNode apply(WorkflowContext workflow, TaskContext<?> task, JsonNode node);
 }
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowInstance.java b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowInstance.java
index bd2f94b..1361c43 100644
--- a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowInstance.java
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowInstance.java
@@ -18,24 +18,26 @@ package io.serverlessworkflow.impl;
 import static io.serverlessworkflow.impl.json.JsonUtils.toJavaValue;
 
 import com.fasterxml.jackson.databind.JsonNode;
-import java.util.Optional;
 
 public class WorkflowInstance {
   private WorkflowState state;
   private WorkflowContext context;
+  private TaskContext<?> taskContext;
 
   WorkflowInstance(WorkflowDefinition definition, JsonNode input) {
     definition.inputSchemaValidator().ifPresent(v -> v.validate(input));
-    context = WorkflowContext.builder(definition, input).build();
+    context = new WorkflowContext(definition, input);
+    taskContext = new TaskContext<>(input, definition.positionFactory().buildPosition());
     definition
         .inputFilter()
-        .ifPresent(f -> context.current(f.apply(context, Optional.empty(), context.current())));
+        .ifPresent(f -> taskContext.input(f.apply(context, taskContext, input)));
     state = WorkflowState.STARTED;
-    WorkflowUtils.processTaskList(definition.workflow().getDo(), context);
+    taskContext.rawOutput(
+        WorkflowUtils.processTaskList(definition.workflow().getDo(), context, taskContext));
     definition
         .outputFilter()
-        .ifPresent(f -> context.current(f.apply(context, Optional.empty(), context.current())));
-    definition.outputSchemaValidator().ifPresent(v -> v.validate(context.current()));
+        .ifPresent(f -> taskContext.output(f.apply(context, taskContext, taskContext.rawOutput())));
+    definition.outputSchemaValidator().ifPresent(v -> v.validate(taskContext.output()));
   }
 
   public WorkflowState state() {
@@ -43,10 +45,10 @@ public class WorkflowInstance {
   }
 
   public Object output() {
-    return toJavaValue(context.current());
+    return toJavaValue(taskContext.output());
   }
 
   public Object outputAsJsonNode() {
-    return context.current();
+    return taskContext.output();
   }
 }
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowPosition.java b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowPosition.java
index c43d4b2..cf63844 100644
--- a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowPosition.java
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowPosition.java
@@ -24,4 +24,6 @@ public interface WorkflowPosition {
   WorkflowPosition addIndex(int index);
 
   WorkflowPosition back();
+
+  WorkflowPosition copy();
 }
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowPositionFactory.java b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowPositionFactory.java
new file mode 100644
index 0000000..e93a4c3
--- /dev/null
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowPositionFactory.java
@@ -0,0 +1,20 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.impl;
+
+public interface WorkflowPositionFactory {
+  WorkflowPosition buildPosition();
+}
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowUtils.java b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowUtils.java
index 259a0e6..54abd7e 100644
--- a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowUtils.java
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowUtils.java
@@ -35,7 +35,6 @@ import io.serverlessworkflow.impl.jsonschema.SchemaValidator;
 import io.serverlessworkflow.impl.jsonschema.SchemaValidatorFactory;
 import io.serverlessworkflow.impl.resources.ResourceLoader;
 import io.serverlessworkflow.impl.resources.StaticResource;
-
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.UncheckedIOException;
@@ -128,32 +127,33 @@ public class WorkflowUtils {
     throw new IllegalArgumentException("Cannot find task with name " + taskName);
   }
 
-  public static void processTaskList(List<TaskItem> tasks, WorkflowContext context) {
-    context.position().addProperty("do");
+  public static JsonNode processTaskList(
+      List<TaskItem> tasks, WorkflowContext context, TaskContext<?> parentTask) {
+    parentTask.position().addProperty("do");
+    TaskContext<? extends TaskBase> currentContext = parentTask;
     if (!tasks.isEmpty()) {
       ListIterator<TaskItem> iter = tasks.listIterator();
       TaskItem nextTask = iter.next();
       while (nextTask != null) {
         TaskItem task = nextTask;
-        context.position().addIndex(iter.nextIndex()).addProperty(task.getName());
+        parentTask.position().addIndex(iter.nextIndex()).addProperty(task.getName());
         context
             .definition()
             .listeners()
-            .forEach(l -> l.onTaskStarted(context.position(), task.getTask()));
-        TaskContext<? extends TaskBase> taskContext =
+            .forEach(l -> l.onTaskStarted(parentTask.position(), task.getTask()));
+        currentContext =
             context
                 .definition()
                 .taskExecutors()
                 .computeIfAbsent(
-                    context.position().jsonPointer(),
+                    parentTask.position().jsonPointer(),
                     k ->
                         context
                             .definition()
                             .taskFactory()
                             .getTaskExecutor(task.getTask(), context.definition()))
-                .apply(context, context.current());
-        context.current(taskContext.output());
-        FlowDirective flowDirective = taskContext.flowDirective();
+                .apply(context, parentTask, currentContext.output());
+        FlowDirective flowDirective = currentContext.flowDirective();
         if (flowDirective.getFlowDirectiveEnum() != null) {
           switch (flowDirective.getFlowDirectiveEnum()) {
             case CONTINUE:
@@ -170,15 +170,21 @@ public class WorkflowUtils {
         context
             .definition()
             .listeners()
-            .forEach(l -> l.onTaskEnded(context.position(), task.getTask()));
-        context.position().back();
+            .forEach(l -> l.onTaskEnded(parentTask.position(), task.getTask()));
+        parentTask.position().back();
       }
     }
-    context.position().back();
+    parentTask.position().back();
+    return currentContext.output();
   }
 
   public static WorkflowFilter buildWorkflowFilter(ExpressionFactory exprFactory, String str) {
+    assert str != null;
     Expression expression = exprFactory.getExpression(str);
     return expression::eval;
   }
+
+  public static Optional<WorkflowFilter> optionalFilter(ExpressionFactory exprFactory, String str) {
+    return str != null ? Optional.of(buildWorkflowFilter(exprFactory, str)) : Optional.empty();
+  }
 }
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/AbstractTaskExecutor.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/AbstractTaskExecutor.java
index 56e5512..086c7c5 100644
--- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/AbstractTaskExecutor.java
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/AbstractTaskExecutor.java
@@ -83,23 +83,20 @@ public abstract class AbstractTaskExecutor<T extends TaskBase> implements TaskEx
   }
 
   @Override
-  public TaskContext<T> apply(WorkflowContext workflowContext, JsonNode rawInput) {
-    TaskContext<T> taskContext = new TaskContext<>(rawInput, task);
+  public TaskContext<T> apply(
+      WorkflowContext workflowContext, TaskContext<?> parentContext, JsonNode input) {
+    TaskContext<T> taskContext = new TaskContext<>(input, parentContext, task);
     inputSchemaValidator.ifPresent(s -> s.validate(taskContext.rawInput()));
     inputProcessor.ifPresent(
-        p ->
-            taskContext.input(
-                p.apply(workflowContext, Optional.of(taskContext), taskContext.rawInput())));
+        p -> taskContext.input(p.apply(workflowContext, taskContext, taskContext.rawInput())));
     internalExecute(workflowContext, taskContext);
     outputProcessor.ifPresent(
-        p ->
-            taskContext.output(
-                p.apply(workflowContext, Optional.of(taskContext), taskContext.rawOutput())));
+        p -> taskContext.output(p.apply(workflowContext, taskContext, taskContext.rawOutput())));
     outputSchemaValidator.ifPresent(s -> s.validate(taskContext.output()));
     contextProcessor.ifPresent(
         p ->
             workflowContext.context(
-                p.apply(workflowContext, Optional.of(taskContext), workflowContext.context())));
+                p.apply(workflowContext, taskContext, workflowContext.context())));
     contextSchemaValidator.ifPresent(s -> s.validate(workflowContext.context()));
     return taskContext;
   }
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/DefaultTaskExecutorFactory.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/DefaultTaskExecutorFactory.java
index b5ac011..45a988a 100644
--- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/DefaultTaskExecutorFactory.java
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/DefaultTaskExecutorFactory.java
@@ -65,6 +65,8 @@ public class DefaultTaskExecutorFactory implements TaskExecutorFactory {
       return new DoExecutor(task.getDoTask(), definition);
     } else if (task.getSetTask() != null) {
       return new SetExecutor(task.getSetTask(), definition);
+    } else if (task.getForTask() != null) {
+      return new ForExecutor(task.getForTask(), definition);
     }
     throw new UnsupportedOperationException(task.get().getClass().getName() + " not supported yet");
   }
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/DoExecutor.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/DoExecutor.java
index 30f35f9..df364c1 100644
--- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/DoExecutor.java
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/DoExecutor.java
@@ -29,6 +29,6 @@ public class DoExecutor extends AbstractTaskExecutor<DoTask> {
 
   @Override
   protected void internalExecute(WorkflowContext workflow, TaskContext<DoTask> taskContext) {
-    WorkflowUtils.processTaskList(task.getDo(), workflow);
+    taskContext.rawOutput(WorkflowUtils.processTaskList(task.getDo(), workflow, taskContext));
   }
 }
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/ForExecutor.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/ForExecutor.java
new file mode 100644
index 0000000..511575f
--- /dev/null
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/ForExecutor.java
@@ -0,0 +1,58 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.impl.executors;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import io.serverlessworkflow.api.types.ForTask;
+import io.serverlessworkflow.api.types.ForTaskConfiguration;
+import io.serverlessworkflow.impl.TaskContext;
+import io.serverlessworkflow.impl.WorkflowContext;
+import io.serverlessworkflow.impl.WorkflowDefinition;
+import io.serverlessworkflow.impl.WorkflowFilter;
+import io.serverlessworkflow.impl.WorkflowUtils;
+import java.util.Iterator;
+import java.util.Optional;
+
+public class ForExecutor extends AbstractTaskExecutor<ForTask> {
+
+  private final WorkflowFilter collectionExpr;
+  private final Optional<WorkflowFilter> whileExpr;
+
+  protected ForExecutor(ForTask task, WorkflowDefinition definition) {
+    super(task, definition);
+    ForTaskConfiguration forConfig = task.getFor();
+    this.collectionExpr =
+        WorkflowUtils.buildWorkflowFilter(definition.expressionFactory(), forConfig.getIn());
+    this.whileExpr = WorkflowUtils.optionalFilter(definition.expressionFactory(), task.getWhile());
+  }
+
+  @Override
+  protected void internalExecute(WorkflowContext workflow, TaskContext<ForTask> taskContext) {
+    Iterator<JsonNode> iter =
+        collectionExpr.apply(workflow, taskContext, taskContext.input()).iterator();
+    int i = 0;
+    while (iter.hasNext()
+        && whileExpr
+            .<JsonNode>map(w -> w.apply(workflow, taskContext, taskContext.rawOutput()))
+            .map(n -> n.asBoolean(true))
+            .orElse(true)) {
+      JsonNode item = iter.next();
+      taskContext.variables().put(task.getFor().getEach(), item);
+      taskContext.variables().put(task.getFor().getAt(), i++);
+      taskContext.rawOutput(WorkflowUtils.processTaskList(task.getDo(), workflow, taskContext));
+    }
+  }
+}
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/SetExecutor.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/SetExecutor.java
index b9d9db8..0f0d999 100644
--- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/SetExecutor.java
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/SetExecutor.java
@@ -15,25 +15,33 @@
  */
 package io.serverlessworkflow.impl.executors;
 
-import com.fasterxml.jackson.databind.JsonNode;
 import io.serverlessworkflow.api.types.SetTask;
 import io.serverlessworkflow.impl.TaskContext;
 import io.serverlessworkflow.impl.WorkflowContext;
 import io.serverlessworkflow.impl.WorkflowDefinition;
+import io.serverlessworkflow.impl.expressions.ExpressionUtils;
 import io.serverlessworkflow.impl.json.JsonUtils;
 import io.serverlessworkflow.impl.json.MergeUtils;
+import java.util.Map;
 
 public class SetExecutor extends AbstractTaskExecutor<SetTask> {
 
-  private JsonNode toBeSet;
+  private Map<String, Object> toBeSet;
 
   protected SetExecutor(SetTask task, WorkflowDefinition definition) {
     super(task, definition);
-    this.toBeSet = JsonUtils.fromValue(task.getSet().getAdditionalProperties());
+    this.toBeSet =
+        ExpressionUtils.buildExpressionMap(
+            task.getSet().getAdditionalProperties(), definition.expressionFactory());
   }
 
   @Override
   protected void internalExecute(WorkflowContext workflow, TaskContext<SetTask> taskContext) {
-    taskContext.rawOutput(MergeUtils.merge(toBeSet, taskContext.input()));
+    taskContext.rawOutput(
+        MergeUtils.merge(
+            JsonUtils.fromValue(
+                ExpressionUtils.evaluateExpressionMap(
+                    toBeSet, workflow, taskContext, taskContext.input())),
+            taskContext.input()));
   }
 }
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/SwitchExecutor.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/SwitchExecutor.java
index f081dfa..dee0cee 100644
--- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/SwitchExecutor.java
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/SwitchExecutor.java
@@ -26,7 +26,6 @@ import io.serverlessworkflow.impl.WorkflowFilter;
 import io.serverlessworkflow.impl.WorkflowUtils;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 
 public class SwitchExecutor extends AbstractTaskExecutor<SwitchTask> {
@@ -52,10 +51,7 @@ public class SwitchExecutor extends AbstractTaskExecutor<SwitchTask> {
   @Override
   protected void internalExecute(WorkflowContext workflow, TaskContext<SwitchTask> taskContext) {
     for (Entry<SwitchCase, WorkflowFilter> entry : workflowFilters.entrySet()) {
-      if (entry
-          .getValue()
-          .apply(workflow, Optional.of(taskContext), taskContext.input())
-          .asBoolean()) {
+      if (entry.getValue().apply(workflow, taskContext, taskContext.input()).asBoolean()) {
         taskContext.flowDirective(entry.getKey().getThen());
         return;
       }
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TaskExecutor.java b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TaskExecutor.java
index 6222819..b4b66a9 100644
--- a/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TaskExecutor.java
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/executors/TaskExecutor.java
@@ -19,7 +19,9 @@ import com.fasterxml.jackson.databind.JsonNode;
 import io.serverlessworkflow.api.types.TaskBase;
 import io.serverlessworkflow.impl.TaskContext;
 import io.serverlessworkflow.impl.WorkflowContext;
-import java.util.function.BiFunction;
 
-public interface TaskExecutor<T extends TaskBase>
-    extends BiFunction<WorkflowContext, JsonNode, TaskContext<T>> {}
+@FunctionalInterface
+public interface TaskExecutor<T extends TaskBase> {
+  TaskContext<T> apply(
+      WorkflowContext workflowContext, TaskContext<?> parentContext, JsonNode input);
+}
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/expressions/Expression.java b/impl/core/src/main/java/io/serverlessworkflow/impl/expressions/Expression.java
index f9d799e..42566c7 100644
--- a/impl/core/src/main/java/io/serverlessworkflow/impl/expressions/Expression.java
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/expressions/Expression.java
@@ -18,8 +18,7 @@ package io.serverlessworkflow.impl.expressions;
 import com.fasterxml.jackson.databind.JsonNode;
 import io.serverlessworkflow.impl.TaskContext;
 import io.serverlessworkflow.impl.WorkflowContext;
-import java.util.Optional;
 
 public interface Expression {
-  JsonNode eval(WorkflowContext workflowContext, Optional<TaskContext<?>> context, JsonNode node);
+  JsonNode eval(WorkflowContext workflowContext, TaskContext<?> context, JsonNode node);
 }
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/expressions/ExpressionUtils.java b/impl/core/src/main/java/io/serverlessworkflow/impl/expressions/ExpressionUtils.java
index 72b4e90..7f77632 100644
--- a/impl/core/src/main/java/io/serverlessworkflow/impl/expressions/ExpressionUtils.java
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/expressions/ExpressionUtils.java
@@ -20,7 +20,6 @@ import io.serverlessworkflow.impl.TaskContext;
 import io.serverlessworkflow.impl.WorkflowContext;
 import io.serverlessworkflow.impl.json.JsonUtils;
 import java.util.Map;
-import java.util.Optional;
 
 public class ExpressionUtils {
 
@@ -35,10 +34,7 @@ public class ExpressionUtils {
   }
 
   public static Map<String, Object> evaluateExpressionMap(
-      Map<String, Object> origMap,
-      WorkflowContext workflow,
-      Optional<TaskContext<?>> task,
-      JsonNode n) {
+      Map<String, Object> origMap, WorkflowContext workflow, TaskContext<?> task, JsonNode n) {
     return new ProxyMap(
         origMap,
         o ->
@@ -54,7 +50,7 @@ public class ExpressionUtils {
   }
 
   public static Object evaluateExpressionObject(
-      Object obj, WorkflowContext workflow, Optional<TaskContext<?>> task, JsonNode node) {
+      Object obj, WorkflowContext workflow, TaskContext<?> task, JsonNode node) {
     return obj instanceof Map
         ? ExpressionUtils.evaluateExpressionMap((Map<String, Object>) obj, workflow, task, node)
         : obj;
diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/expressions/JQExpression.java b/impl/core/src/main/java/io/serverlessworkflow/impl/expressions/JQExpression.java
index 7b008fc..ae6c784 100644
--- a/impl/core/src/main/java/io/serverlessworkflow/impl/expressions/JQExpression.java
+++ b/impl/core/src/main/java/io/serverlessworkflow/impl/expressions/JQExpression.java
@@ -17,6 +17,7 @@ package io.serverlessworkflow.impl.expressions;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.ArrayNode;
+import io.serverlessworkflow.impl.ContextAware;
 import io.serverlessworkflow.impl.TaskContext;
 import io.serverlessworkflow.impl.WorkflowContext;
 import io.serverlessworkflow.impl.json.JsonUtils;
@@ -177,10 +178,10 @@ public class JQExpression implements Expression {
   }
 
   @Override
-  public JsonNode eval(WorkflowContext workflow, Optional<TaskContext<?>> task, JsonNode node) {
+  public JsonNode eval(WorkflowContext workflow, TaskContext<?> task, JsonNode node) {
     TypedOutput<JsonNode> output = output(JsonNode.class);
     try {
-      internalExpr.apply(this.scope.get(), node, output);
+      internalExpr.apply(createScope(workflow, task), node, output);
       return output.getResult();
     } catch (JsonQueryException e) {
       throw new IllegalArgumentException(
@@ -188,6 +189,16 @@ public class JQExpression implements Expression {
     }
   }
 
+  private Scope createScope(WorkflowContext workflow, TaskContext<?> task) {
+    return createScope(scope.get(), task);
+  }
+
+  private Scope createScope(Scope parentScope, ContextAware context) {
+    Scope childScope = Scope.newChildScope(parentScope);
+    context.variables().forEach((k, v) -> childScope.setValue(k, JsonUtils.fromValue(v)));
+    return childScope;
+  }
+
   private void checkFunctionCall(net.thisptr.jackson.jq.Expression toCheck)
       throws JsonQueryException {
     if (toCheck instanceof FunctionCall) {
diff --git a/impl/http/src/main/java/io/serverlessworkflow/impl/executors/HttpExecutor.java b/impl/http/src/main/java/io/serverlessworkflow/impl/executors/HttpExecutor.java
index 684acd4..5ecd27a 100644
--- a/impl/http/src/main/java/io/serverlessworkflow/impl/executors/HttpExecutor.java
+++ b/impl/http/src/main/java/io/serverlessworkflow/impl/executors/HttpExecutor.java
@@ -38,7 +38,6 @@ import jakarta.ws.rs.client.Invocation.Builder;
 import jakarta.ws.rs.client.WebTarget;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.Optional;
 
 public class HttpExecutor implements CallableTask<CallHTTP> {
 
@@ -82,8 +81,7 @@ public class HttpExecutor implements CallableTask<CallHTTP> {
             (request, workflow, context, node) ->
                 request.post(
                     Entity.json(
-                        ExpressionUtils.evaluateExpressionObject(
-                            body, workflow, Optional.of(context), node)),
+                        ExpressionUtils.evaluateExpressionObject(body, workflow, context, node)),
                     JsonNode.class);
         break;
       case HttpMethod.GET:
@@ -97,12 +95,11 @@ public class HttpExecutor implements CallableTask<CallHTTP> {
       WorkflowContext workflow, TaskContext<CallHTTP> taskContext, JsonNode input) {
     WebTarget target = targetSupplier.apply(workflow, taskContext, input);
     for (Entry<String, Object> entry :
-        ExpressionUtils.evaluateExpressionMap(queryMap, workflow, Optional.of(taskContext), input)
-            .entrySet()) {
+        ExpressionUtils.evaluateExpressionMap(queryMap, workflow, taskContext, input).entrySet()) {
       target = target.queryParam(entry.getKey(), entry.getValue());
     }
     Builder request = target.request();
-    ExpressionUtils.evaluateExpressionMap(headersMap, workflow, Optional.of(taskContext), input)
+    ExpressionUtils.evaluateExpressionMap(headersMap, workflow, taskContext, input)
         .forEach(request::header);
     return requestFunction.apply(request, workflow, taskContext, input);
   }
@@ -153,7 +150,7 @@ public class HttpExecutor implements CallableTask<CallHTTP> {
 
     @Override
     public WebTarget apply(WorkflowContext workflow, TaskContext<?> task, JsonNode node) {
-      return client.target(expr.eval(workflow, Optional.of(task), node).asText());
+      return client.target(expr.eval(workflow, task, node).asText());
     }
   }
 }


Test patch:
diff --git a/impl/core/src/test/java/io/serverlessworkflow/impl/WorkflowDefinitionTest.java b/impl/core/src/test/java/io/serverlessworkflow/impl/WorkflowDefinitionTest.java
index f6c3455..39adcf9 100644
--- a/impl/core/src/test/java/io/serverlessworkflow/impl/WorkflowDefinitionTest.java
+++ b/impl/core/src/test/java/io/serverlessworkflow/impl/WorkflowDefinitionTest.java
@@ -19,6 +19,7 @@ import static io.serverlessworkflow.api.WorkflowReader.readWorkflowFromClasspath
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.Map;
 import java.util.stream.Stream;
 import org.assertj.core.api.Condition;
@@ -45,10 +46,6 @@ public class WorkflowDefinitionTest {
   }
 
   private static Stream<Arguments> provideParameters() {
-    Map<String, Object> petInput = Map.of("petId", 10);
-    Condition<Object> petCondition =
-        new Condition<>(
-            o -> ((Map<String, Object>) o).containsKey("photoUrls"), "callHttpCondition");
     return Stream.of(
         Arguments.of(
             "switch-then-string.yaml",
@@ -82,6 +79,18 @@ public class WorkflowDefinitionTest {
                     o.equals(
                         Map.of(
                             "orderType", "unknown", "log", "warn", "message", "something's wrong")),
-                "switch-unknown")));
+                "switch-unknown")),
+        Arguments.of(
+            "for-sum.yaml",
+            Map.of("input", Arrays.asList(1, 2, 3)),
+            new Condition(o -> o.equals(6), "for-sum")),
+        Arguments.of(
+            "for-collect.yaml",
+            Map.of("input", Arrays.asList(1, 2, 3)),
+            new Condition(
+                o ->
+                    o.equals(
+                        Map.of("input", Arrays.asList(1, 2, 3), "output", Arrays.asList(2, 4, 6))),
+                "for-collect")));
   }
 }
diff --git a/impl/core/src/test/resources/for-collect.yaml b/impl/core/src/test/resources/for-collect.yaml
new file mode 100644
index 0000000..7bcc48c
--- /dev/null
+++ b/impl/core/src/test/resources/for-collect.yaml
@@ -0,0 +1,17 @@
+document:
+  dsl: '1.0.0-alpha5'
+  namespace: test
+  name: for-collect-example
+  version: '0.1.0'
+do:
+  - sumAll:
+      for:
+        each: number
+        in: .input
+        at: index
+      input: 
+        from: '{input: .input, output: []}'
+      do:
+        - sumIndex:
+            output:
+              as: .output+=[$number+$index+1]
\ No newline at end of file
diff --git a/impl/core/src/test/resources/for-sum.yaml b/impl/core/src/test/resources/for-sum.yaml
new file mode 100644
index 0000000..e0fe106
--- /dev/null
+++ b/impl/core/src/test/resources/for-sum.yaml
@@ -0,0 +1,19 @@
+document:
+  dsl: '1.0.0-alpha5'
+  namespace: test
+  name: for-sum-example
+  version: '0.1.0'
+do:
+  - initCounter:
+      set:  
+       counter: 0
+  - sumAll:
+      for:
+        each: number
+        in: .input
+      do:
+        - accumulate:
+            output:
+              as: .counter+=$number
+      output: 
+       as: .counter
\ No newline at end of file