diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/ConcordJsonSchemaGenerator.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/ConcordJsonSchemaGenerator.java
index eec231a4a5..f2bce4c96b 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/ConcordJsonSchemaGenerator.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/ConcordJsonSchemaGenerator.java
@@ -21,11 +21,7 @@
  */
 
 import com.fasterxml.jackson.core.JsonGenerator;
-import com.fasterxml.jackson.databind.JavaType;
-import com.fasterxml.jackson.databind.JsonMappingException;
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.*;
 import com.fasterxml.jackson.databind.jsonFormatVisitors.JsonFormatVisitorWrapper;
 import com.fasterxml.jackson.databind.module.SimpleModule;
 import com.fasterxml.jackson.databind.node.ObjectNode;
@@ -35,16 +31,8 @@
 import com.kjetland.jackson.jsonSchema.SubclassesResolver;
 import com.kjetland.jackson.jsonSchema.SubclassesResolverImpl;
 import com.walmartlabs.concord.imports.Imports;
-import com.walmartlabs.concord.runtime.v2.model.Form;
-import com.walmartlabs.concord.runtime.v2.model.ProcessDefinition;
-import com.walmartlabs.concord.runtime.v2.model.ProcessDefinitionConfiguration;
-import com.walmartlabs.concord.runtime.v2.model.Step;
-import com.walmartlabs.concord.runtime.v2.model.Trigger;
-import com.walmartlabs.concord.runtime.v2.schema.ImportsMixIn;
-import com.walmartlabs.concord.runtime.v2.schema.ProcessDefinitionConfigurationMixIn;
-import com.walmartlabs.concord.runtime.v2.schema.ProcessDefinitionMixIn;
-import com.walmartlabs.concord.runtime.v2.schema.StepMixIn;
-import com.walmartlabs.concord.runtime.v2.schema.TriggerMixIn;
+import com.walmartlabs.concord.runtime.v2.model.*;
+import com.walmartlabs.concord.runtime.v2.schema.*;
 
 import java.io.IOException;
 import java.io.OutputStream;
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/ProjectSerializerV2.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/ProjectSerializerV2.java
index cb63eaa48b..29e56b9e9d 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/ProjectSerializerV2.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/ProjectSerializerV2.java
@@ -56,6 +56,7 @@ private static ObjectMapper createObjectMapper() {
         SimpleModule module = new SimpleModule()
                 .addSerializer(SimpleOptions.class, new SimpleOptionsSerializer())
                 .addSerializer(Retry.class, new RetryOptionsSerializer())
+                .addSerializer(Loop.class, new LoopOptionsSerializer())
                 .addSerializer(WithItems.class, new WithItemsSerializer())
                 .addSerializer(Checkpoint.class, new CheckpointStepSerializer())
                 .addSerializer(ExitStep.class, new ExitStepSerializer())
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/FlowCallOptions.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/FlowCallOptions.java
index 6c7aaf1c73..7f751bb379 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/FlowCallOptions.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/FlowCallOptions.java
@@ -59,6 +59,9 @@ default Map<String, Serializable> outExpr() {
     @Nullable
     WithItems withItems();
 
+    @Nullable
+    Loop loop();
+
     @Nullable
     Retry retry();
 
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/GroupOfStepsOptions.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/GroupOfStepsOptions.java
index 6c4680edd2..9ed2f35894 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/GroupOfStepsOptions.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/GroupOfStepsOptions.java
@@ -46,6 +46,9 @@ default List<Step> errorSteps() {
     @Nullable
     WithItems withItems();
 
+    @Nullable
+    Loop loop();
+
     static ImmutableGroupOfStepsOptions.Builder builder() {
         return ImmutableGroupOfStepsOptions.builder();
     }
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/Loop.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/Loop.java
new file mode 100644
index 0000000000..682c77f4c8
--- /dev/null
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/Loop.java
@@ -0,0 +1,55 @@
+package com.walmartlabs.concord.runtime.v2.model;
+
+/*-
+ * *****
+ * Concord
+ * -----
+ * Copyright (C) 2017 - 2019 Walmart Inc.
+ * -----
+ * 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.
+ * =====
+ */
+
+import org.immutables.value.Value;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.Map;
+
+@Value.Immutable
+@Value.Style(jdkOnly = true)
+public interface Loop extends Serializable {
+
+    long serialVersionUID = 1L;
+
+    static ImmutableLoop.Builder builder() {
+        return  ImmutableLoop.builder();
+    }
+
+    Serializable items();
+
+    @Value.Default
+    default Mode mode() {
+        return Mode.SERIAL;
+    }
+
+    @Value.Default
+    default Map<String, Object> options() {
+        return Collections.emptyMap();
+    }
+
+    enum Mode {
+        SERIAL,
+        PARALLEL
+    }
+}
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/ProcessDefinition.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/ProcessDefinition.java
index 70b9f43393..6afebf6617 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/ProcessDefinition.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/ProcessDefinition.java
@@ -24,7 +24,10 @@
 import org.immutables.value.Value;
 
 import java.io.Serializable;
-import java.util.*;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
 
 @Value.Immutable
 @Value.Style(jdkOnly = true)
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/ProcessDefinitionConfiguration.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/ProcessDefinitionConfiguration.java
index 5b197a3791..65297793f5 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/ProcessDefinitionConfiguration.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/ProcessDefinitionConfiguration.java
@@ -106,6 +106,11 @@ default List<String> out() {
     @Nullable
     String template();
 
+    @Value.Default
+    default int parallelLoopParallelism() {
+        return Runtime.getRuntime().availableProcessors();
+    }
+
     static ImmutableProcessDefinitionConfiguration.Builder builder() {
         return ImmutableProcessDefinitionConfiguration.builder();
     }
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/ScriptCallOptions.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/ScriptCallOptions.java
index d4aaa914b4..03cd24b4d3 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/ScriptCallOptions.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/ScriptCallOptions.java
@@ -60,6 +60,9 @@ default Map<String, Serializable> outExpr() {
     @Nullable
     WithItems withItems();
 
+    @Nullable
+    Loop loop();
+
     @Nullable
     Retry retry();
 
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/TaskCallOptions.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/TaskCallOptions.java
index eda223747e..37b785dfb1 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/TaskCallOptions.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/TaskCallOptions.java
@@ -57,6 +57,9 @@ default Map<String, Serializable> outExpr() {
     @Nullable
     WithItems withItems();
 
+    @Nullable
+    Loop loop();
+
     @Nullable
     Retry retry();
 
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/WithItems.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/WithItems.java
index 40f6def207..052e245703 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/WithItems.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/model/WithItems.java
@@ -24,6 +24,10 @@
 
 import java.io.Serializable;
 
+/**
+ * @deprecated use {@link com.walmartlabs.concord.runtime.v2.model.Loop}
+ */
+@Deprecated
 @Value.Immutable
 @Value.Style(jdkOnly = true)
 public interface WithItems extends Serializable {
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/ConfigurationGrammar.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/ConfigurationGrammar.java
index 7883ffc0ba..963be41c44 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/ConfigurationGrammar.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/ConfigurationGrammar.java
@@ -84,7 +84,8 @@ public final class ConfigurationGrammar {
                                     optional("out", stringArrayVal.map(o::addAllOut)),
                                     optional("arguments", mapVal.map(o::arguments)),
                                     optional("debug", booleanVal.map(o::debug)),
-                                    optional("template", stringVal.map(o::template))))
+                                    optional("template", stringVal.map(o::template)),
+                                    optional("parallelLoopParallelism", intVal.map(o::parallelLoopParallelism))))
                             .map(ImmutableProcessDefinitionConfiguration.Builder::build));
 
     public static final Parser<Atom, ProcessDefinitionConfiguration> processCfgVal =
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/ExpressionGrammar.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/ExpressionGrammar.java
index 7e5fae58df..e3e076161a 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/ExpressionGrammar.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/ExpressionGrammar.java
@@ -22,7 +22,10 @@
 
 import com.fasterxml.jackson.core.JsonToken;
 import com.walmartlabs.concord.runtime.v2.Constants;
-import com.walmartlabs.concord.runtime.v2.model.*;
+import com.walmartlabs.concord.runtime.v2.model.Expression;
+import com.walmartlabs.concord.runtime.v2.model.ExpressionOptions;
+import com.walmartlabs.concord.runtime.v2.model.ImmutableExpressionOptions;
+import com.walmartlabs.concord.runtime.v2.model.Step;
 import io.takari.parc.Parser;
 
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarMisc.*;
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/FlowCallGrammar.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/FlowCallGrammar.java
index cbc3398ddd..c1d319216e 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/FlowCallGrammar.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/FlowCallGrammar.java
@@ -32,6 +32,7 @@
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarOptions.optional;
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarOptions.options;
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarV2.*;
+import static com.walmartlabs.concord.runtime.v2.parser.LoopGrammar.loopVal;
 import static com.walmartlabs.concord.runtime.v2.parser.RetryGrammar.retryVal;
 import static io.takari.parc.Combinators.or;
 
@@ -54,6 +55,7 @@ private static Parser<Atom, FlowCallOptions> callOptions(String stepName) {
                         optional("name", stringVal.map(v -> o.putMeta(Constants.SEGMENT_NAME, v))),
                         optional("withItems", nonNullVal.map(v -> o.withItems(WithItems.of(v, WithItems.Mode.SERIAL)))),
                         optional("parallelWithItems", nonNullVal.map(v -> o.withItems(WithItems.of(v, WithItems.Mode.PARALLEL)))),
+                        optional("loop", loopVal.map(o::loop)),
                         optional("retry", retryVal.map(o::retry)),
                         optional("error", stepsVal.map(o::errorSteps))
                 ))
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/GrammarV2.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/GrammarV2.java
index 38b2e31c26..78d2ece4d9 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/GrammarV2.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/GrammarV2.java
@@ -31,6 +31,7 @@
 import java.time.Duration;
 import java.time.format.DateTimeParseException;
 import java.util.*;
+import java.util.function.BiPredicate;
 import java.util.regex.Pattern;
 import java.util.regex.PatternSyntaxException;
 import java.util.stream.Collectors;
@@ -96,12 +97,17 @@ public final class GrammarV2 {
     });
 
     public static <E extends Enum<E>> Parser<Atom, E> enumVal(Class<E> enumData) {
+        return enumVal(enumData, String::equals);
+    }
+
+    public static <E extends Enum<E>> Parser<Atom, E> enumVal(Class<E> enumData,
+                                                              BiPredicate<String, String> cmp) {
         return value.map(vv -> {
             String v = vv.getValue(YamlValueType.STRING);
 
-            for (Enum<E> enumVal : enumData.getEnumConstants()) {
-                if (enumVal.name().equals(v)) {
-                    return Enum.valueOf(enumData, v);
+            for (E enumVal : enumData.getEnumConstants()) {
+                if (cmp.test(enumVal.name(), v)) {
+                    return enumVal;
                 }
             }
 
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/GroupOfStepsGrammar.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/GroupOfStepsGrammar.java
index cb99d67785..faed780db9 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/GroupOfStepsGrammar.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/GroupOfStepsGrammar.java
@@ -24,10 +24,12 @@
 import com.walmartlabs.concord.runtime.v2.model.*;
 import io.takari.parc.Parser;
 
-import static com.walmartlabs.concord.runtime.v2.parser.GrammarMisc.*;
+import static com.walmartlabs.concord.runtime.v2.parser.GrammarMisc.namedStep;
+import static com.walmartlabs.concord.runtime.v2.parser.GrammarMisc.with;
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarOptions.optional;
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarOptions.options;
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarV2.*;
+import static com.walmartlabs.concord.runtime.v2.parser.LoopGrammar.loopVal;
 import static io.takari.parc.Combinators.choice;
 
 public final class GroupOfStepsGrammar {
@@ -47,6 +49,7 @@ private static Parser<Atom, GroupOfStepsOptions> groupOptions(String stepName) {
                         optional("error", stepsVal.map(o::errorSteps)),
                         optional("withItems", nonNullVal.map(v -> o.withItems(WithItems.of(v, WithItems.Mode.SERIAL)))),
                         optional("parallelWithItems", nonNullVal.map(v -> o.withItems(WithItems.of(v, WithItems.Mode.PARALLEL)))),
+                        optional("loop", loopVal.map(o::loop)),
                         optional("meta", mapVal.map(o::putAllMeta)),
                         optional("name", stringVal.map(v -> o.putMeta(Constants.SEGMENT_NAME, v)))
                 ))
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/LoopGrammar.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/LoopGrammar.java
new file mode 100644
index 0000000000..6bdc6e46cb
--- /dev/null
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/LoopGrammar.java
@@ -0,0 +1,49 @@
+package com.walmartlabs.concord.runtime.v2.parser;
+
+/*-
+ * *****
+ * Concord
+ * -----
+ * Copyright (C) 2017 - 2019 Walmart Inc.
+ * -----
+ * 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.
+ * =====
+ */
+
+import com.fasterxml.jackson.core.JsonToken;
+import com.walmartlabs.concord.runtime.v2.model.ImmutableLoop;
+import com.walmartlabs.concord.runtime.v2.model.Loop;
+import io.takari.parc.Parser;
+
+import static com.walmartlabs.concord.runtime.v2.parser.GrammarMisc.*;
+import static com.walmartlabs.concord.runtime.v2.parser.GrammarOptions.*;
+import static com.walmartlabs.concord.runtime.v2.parser.GrammarV2.*;
+
+public final class LoopGrammar {
+
+    private static final Parser<Atom, Loop> loop =
+            betweenTokens(JsonToken.START_OBJECT, JsonToken.END_OBJECT,
+                    with(Loop::builder,
+                            o -> options(
+                                    mandatory("items", nonNullVal.map(o::items)),
+                                    optional("mode", enumVal(Loop.Mode.class, String::equalsIgnoreCase).map(o::mode)),
+                                    optional("parallelism", intVal.map(v -> o.putOptions("parallelism", v)))
+                            ))
+                            .map(ImmutableLoop.Builder::build));
+
+    public static final Parser<Atom, Loop> loopVal =
+            orError(loop, YamlValueType.LOOP);
+
+    private LoopGrammar() {
+    }
+}
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/ParallelGrammar.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/ParallelGrammar.java
index 2d62261d79..6db67b2b1d 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/ParallelGrammar.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/ParallelGrammar.java
@@ -21,7 +21,6 @@
  */
 
 import com.walmartlabs.concord.runtime.v2.model.ImmutableParallelBlockOptions;
-import com.walmartlabs.concord.runtime.v2.model.ImmutableTaskCallOptions;
 import com.walmartlabs.concord.runtime.v2.model.ParallelBlock;
 import com.walmartlabs.concord.runtime.v2.model.ParallelBlockOptions;
 import io.takari.parc.Parser;
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/ScriptGrammar.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/ScriptGrammar.java
index ce1b65f83b..0c84135c24 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/ScriptGrammar.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/ScriptGrammar.java
@@ -21,7 +21,10 @@
  */
 
 import com.walmartlabs.concord.runtime.v2.Constants;
-import com.walmartlabs.concord.runtime.v2.model.*;
+import com.walmartlabs.concord.runtime.v2.model.ImmutableScriptCallOptions;
+import com.walmartlabs.concord.runtime.v2.model.ScriptCall;
+import com.walmartlabs.concord.runtime.v2.model.ScriptCallOptions;
+import com.walmartlabs.concord.runtime.v2.model.WithItems;
 import io.takari.parc.Parser;
 
 import static com.walmartlabs.concord.runtime.v2.parser.ExpressionGrammar.maybeExpression;
@@ -29,6 +32,7 @@
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarOptions.optional;
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarOptions.options;
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarV2.*;
+import static com.walmartlabs.concord.runtime.v2.parser.LoopGrammar.loopVal;
 import static com.walmartlabs.concord.runtime.v2.parser.RetryGrammar.retryVal;
 import static io.takari.parc.Combinators.or;
 
@@ -52,6 +56,7 @@ private static Parser<Atom, ScriptCallOptions> scriptOptions(String stepName) {
                         optional("name", stringVal.map(v -> o.putMeta(Constants.SEGMENT_NAME, v))),
                         optional("withItems", nonNullVal.map(v -> o.withItems(WithItems.of(v, WithItems.Mode.SERIAL)))),
                         optional("parallelWithItems", nonNullVal.map(v -> o.withItems(WithItems.of(v, WithItems.Mode.PARALLEL)))),
+                        optional("loop", loopVal.map(o::loop)),
                         optional("retry", retryVal.map(o::retry)),
                         optional("error", stepsVal.map(o::errorSteps))
                 ))
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/TaskGrammar.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/TaskGrammar.java
index f1341fec36..ba262a0a59 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/TaskGrammar.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/TaskGrammar.java
@@ -32,6 +32,7 @@
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarOptions.optional;
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarOptions.options;
 import static com.walmartlabs.concord.runtime.v2.parser.GrammarV2.*;
+import static com.walmartlabs.concord.runtime.v2.parser.LoopGrammar.loopVal;
 import static com.walmartlabs.concord.runtime.v2.parser.RetryGrammar.retryVal;
 import static io.takari.parc.Combinators.or;
 
@@ -62,6 +63,7 @@ private static Parser<Atom, TaskCallOptions> taskOptions(String stepName) {
                         optional("name", stringVal.map(v -> o.putMeta(Constants.SEGMENT_NAME, v))),
                         optional("withItems", nonNullVal.map(v -> o.withItems(WithItems.of(v, WithItems.Mode.SERIAL)))),
                         optional("parallelWithItems", nonNullVal.map(v -> o.withItems(WithItems.of(v, WithItems.Mode.PARALLEL)))),
+                        optional("loop", loopVal.map(o::loop)),
                         optional("retry", retryVal.map(o::retry)),
                         optional("error", stepsVal.map(o::errorSteps)),
                         optional("ignoreErrors", booleanVal.map(o::ignoreErrors))
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/YamlValueType.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/YamlValueType.java
index 4f611e895f..ec5ccf3ae1 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/YamlValueType.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/parser/YamlValueType.java
@@ -108,6 +108,7 @@ public final class YamlValueType<T> {
     public static final YamlValueType<GithubTriggerExclusiveMode> GITHUB_EXCLUSIVE_MODE = type("GITHUB_EXCLUSIVE_MODE");
     public static final YamlValueType<Map<String, Object>> GITHUB_REPOSITORY_INFO = type("GITHUB_REPOSITORY_INFO");
     public static final YamlValueType<List<Map<String, Object>>> ARRAY_OF_GITHUB_REPOSITORY_INFO = array("REPOSITORY_INFO", GITHUB_REPOSITORY_INFO);
+    public static final YamlValueType<Loop> LOOP = type("LOOP");
 
     private final String name;
 
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/schema/FlowCallStepMixIn.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/schema/FlowCallStepMixIn.java
index f9a3ae6213..d21af51121 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/schema/FlowCallStepMixIn.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/schema/FlowCallStepMixIn.java
@@ -50,11 +50,8 @@ public interface FlowCallStepMixIn extends NamedStep {
     @JsonProperty("error")
     List<Step> error();
 
-    @JsonProperty("withItems")
-    WithItemsMixIn withItems();
-
-    @JsonProperty("parallelWithItems")
-    WithItemsMixIn parallelWithItems();
+    @JsonProperty("loop")
+    LoopMixIn loop();
 
     @JsonProperty("meta")
     Map<String, Object> meta();
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/schema/GroupOfStepsMixIn.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/schema/GroupOfStepsMixIn.java
index 34efac440a..4f549de9ca 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/schema/GroupOfStepsMixIn.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/schema/GroupOfStepsMixIn.java
@@ -34,11 +34,8 @@ public interface GroupOfStepsMixIn extends NamedStep {
     @JsonSchemaInject(json = "{\"oneOf\": [ {\"type\": \"array\", \"items\" : {\"type\" : \"string\"}}, {\"type\": \"string\"} ]}", merge = false)
     Object out();
 
-    @JsonProperty("withItems")
-    WithItemsMixIn withItems();
-
-    @JsonProperty("parallelWithItems")
-    WithItemsMixIn parallelWithItems();
+    @JsonProperty("loop")
+    LoopMixIn loop();
 
     @JsonProperty("error")
     List<StepMixIn> errorSteps();
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/schema/WithItemsMixIn.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/schema/LoopMixIn.java
similarity index 64%
rename from runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/schema/WithItemsMixIn.java
rename to runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/schema/LoopMixIn.java
index c68927aad8..d0a7cda02c 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/schema/WithItemsMixIn.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/schema/LoopMixIn.java
@@ -20,10 +20,21 @@
  * =====
  */
 
+import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonTypeName;
 import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaInject;
+import com.walmartlabs.concord.runtime.v2.model.Loop;
 
-@JsonTypeName("WithItems")
-@JsonSchemaInject(json = "{\"type\": [\"string\", \"object\", \"array\"]}", merge = false)
-public interface WithItemsMixIn {
+@JsonTypeName("Loop")
+public interface LoopMixIn {
+
+    @JsonProperty("items")
+    @JsonSchemaInject(json = "{\"oneOf\": [ {\"type\": \"string\"}, {\"type\": \"object\"}, {\"type\": \"array\"} ]}", merge = false)
+    Object items();
+
+    @JsonProperty("mode")
+    Loop.Mode mode();
+
+    @JsonProperty("parallelism")
+    Integer parallelism();
 }
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/schema/ScriptCallMixIn.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/schema/ScriptCallMixIn.java
index 4837f9c69f..6e3b3d0373 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/schema/ScriptCallMixIn.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/schema/ScriptCallMixIn.java
@@ -25,7 +25,6 @@
 import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaInject;
 import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle;
 
-import java.io.Serializable;
 import java.util.List;
 import java.util.Map;
 
@@ -47,11 +46,8 @@ public interface ScriptCallMixIn extends StepMixIn {
     @JsonSchemaInject(json = "{\"oneOf\": [ {\"type\": \"string\"}, {\"type\": \"object\"} ]}", merge = false)
     Object out();
 
-    @JsonProperty("withItems")
-    WithItemsMixIn withItems();
-
-    @JsonProperty("parallelWithItems")
-    WithItemsMixIn parallelWithItems();
+    @JsonProperty("loop")
+    LoopMixIn loop();
 
     @JsonProperty("retry")
     RetryMixIn retry();
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/schema/TaskCallMixIn.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/schema/TaskCallMixIn.java
index a2b96601dc..d7231c0530 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/schema/TaskCallMixIn.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/schema/TaskCallMixIn.java
@@ -41,11 +41,8 @@ public interface TaskCallMixIn extends NamedStep {
     @JsonSchemaInject(json = "{\"oneOf\": [ {\"type\": \"string\"}, {\"type\": \"object\"} ]}", merge = false)
     Object out();
 
-    @JsonProperty("withItems")
-    WithItemsMixIn withItems();
-
-    @JsonProperty("parallelWithItems")
-    WithItemsMixIn parallelWithItems();
+    @JsonProperty("loop")
+    LoopMixIn loop();
 
     @JsonProperty("retry")
     RetryMixIn retry();
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/serializer/DurationSerializer.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/serializer/DurationSerializer.java
index f0b747ba83..ac9deb8ed0 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/serializer/DurationSerializer.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/serializer/DurationSerializer.java
@@ -23,14 +23,9 @@
 import com.fasterxml.jackson.core.JsonGenerator;
 import com.fasterxml.jackson.databind.SerializerProvider;
 import com.fasterxml.jackson.databind.ser.std.StdSerializer;
-import com.walmartlabs.concord.runtime.v2.model.ProcessDefinition;
 
 import java.io.IOException;
 import java.time.Duration;
-import java.util.Collections;
-import java.util.stream.Collectors;
-
-import static com.walmartlabs.concord.runtime.v2.serializer.SerializerUtils.writeNotEmptyObjectField;
 
 public class DurationSerializer extends StdSerializer<Duration> {
 
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/serializer/FlowCallStepSerializer.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/serializer/FlowCallStepSerializer.java
index 00d15064bf..5a211efbf8 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/serializer/FlowCallStepSerializer.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/serializer/FlowCallStepSerializer.java
@@ -30,6 +30,7 @@
 import java.io.IOException;
 import java.util.Objects;
 
+import static com.walmartlabs.concord.runtime.v2.serializer.SerializerUtils.writeLoop;
 import static com.walmartlabs.concord.runtime.v2.serializer.SerializerUtils.writeNotEmptyObjectField;
 
 public class FlowCallStepSerializer extends StdSerializer<FlowCall> {
@@ -70,6 +71,8 @@ private static void serializeOptions(FlowCallOptions options, JsonGenerator gen)
             SerializerUtils.writeWithItems(items, gen);
         }
 
+        writeLoop(options.loop(), gen);
+
         if (options.retry() != null) {
             gen.writeObjectField("retry", options.retry());
         }
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/serializer/LoopOptionsSerializer.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/serializer/LoopOptionsSerializer.java
new file mode 100644
index 0000000000..88b00ecc17
--- /dev/null
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/serializer/LoopOptionsSerializer.java
@@ -0,0 +1,56 @@
+package com.walmartlabs.concord.runtime.v2.serializer;
+
+/*-
+ * *****
+ * Concord
+ * -----
+ * Copyright (C) 2017 - 2020 Walmart Inc.
+ * -----
+ * 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.
+ * =====
+ */
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.ser.std.StdSerializer;
+import com.walmartlabs.concord.runtime.v2.model.Loop;
+
+import java.io.IOException;
+import java.util.Map;
+
+public class LoopOptionsSerializer extends StdSerializer<Loop> {
+
+    private static final long serialVersionUID = 1L;
+
+    public LoopOptionsSerializer() {
+        this(null);
+    }
+
+    public LoopOptionsSerializer(Class<Loop> t) {
+        super(t);
+    }
+
+    @Override
+    public void serialize(Loop value, JsonGenerator gen, SerializerProvider provider) throws IOException {
+        gen.writeStartObject();
+
+        gen.writeObjectField("mode", value.mode().name().toLowerCase());
+        gen.writeObjectField("items", value.items());
+
+        for (Map.Entry<String, Object> e : value.options().entrySet()) {
+            gen.writeObjectField(e.getKey(), e.getValue());
+        }
+
+        gen.writeEndObject();
+    }
+}
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/serializer/ScriptCallStepSerializer.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/serializer/ScriptCallStepSerializer.java
index b8d99fd0ea..bbec689fb2 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/serializer/ScriptCallStepSerializer.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/serializer/ScriptCallStepSerializer.java
@@ -30,8 +30,7 @@
 import java.io.IOException;
 import java.util.Objects;
 
-import static com.walmartlabs.concord.runtime.v2.serializer.SerializerUtils.writeNotEmptyObjectField;
-import static com.walmartlabs.concord.runtime.v2.serializer.SerializerUtils.writeWithItems;
+import static com.walmartlabs.concord.runtime.v2.serializer.SerializerUtils.*;
 
 public class ScriptCallStepSerializer extends StdSerializer<ScriptCall> {
 
@@ -75,6 +74,8 @@ private static void serializeOptions(ScriptCallOptions options, JsonGenerator ge
             writeWithItems(items, gen);
         }
 
+        writeLoop(options.loop(), gen);
+
         if (options.retry() != null) {
             gen.writeObjectField("retry", options.retry());
         }
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/serializer/SerializerUtils.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/serializer/SerializerUtils.java
index 1278ae9832..ce9b217a39 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/serializer/SerializerUtils.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/serializer/SerializerUtils.java
@@ -21,6 +21,7 @@
  */
 
 import com.fasterxml.jackson.core.JsonGenerator;
+import com.walmartlabs.concord.runtime.v2.model.Loop;
 import com.walmartlabs.concord.runtime.v2.model.WithItems;
 
 import java.io.IOException;
@@ -53,6 +54,7 @@ public static <T> void writeNotEmptyObjectField(String fieldName, Collection<T>
         gen.writeObjectField(fieldName, value);
     }
 
+    @Deprecated
     public static void writeWithItems(WithItems items, JsonGenerator gen) throws IOException {
         switch (items.mode()) {
             case PARALLEL: {
@@ -68,6 +70,14 @@ public static void writeWithItems(WithItems items, JsonGenerator gen) throws IOE
         }
     }
 
+    public static void writeLoop(Loop loop, JsonGenerator gen) throws IOException {
+        if (loop == null) {
+            return;
+        }
+
+        gen.writeObjectField("loop", loop);
+    }
+
     private SerializerUtils() {
     }
 }
diff --git a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/serializer/TaskCallStepSerializer.java b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/serializer/TaskCallStepSerializer.java
index 2469e1ea54..9544fc392a 100644
--- a/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/serializer/TaskCallStepSerializer.java
+++ b/runtime/v2/model/src/main/java/com/walmartlabs/concord/runtime/v2/serializer/TaskCallStepSerializer.java
@@ -30,8 +30,7 @@
 import java.io.IOException;
 import java.util.Objects;
 
-import static com.walmartlabs.concord.runtime.v2.serializer.SerializerUtils.writeNotEmptyObjectField;
-import static com.walmartlabs.concord.runtime.v2.serializer.SerializerUtils.writeWithItems;
+import static com.walmartlabs.concord.runtime.v2.serializer.SerializerUtils.*;
 
 public class TaskCallStepSerializer extends StdSerializer<TaskCall> {
 
@@ -69,6 +68,8 @@ public void serialize(TaskCall value, JsonGenerator gen, SerializerProvider prov
             writeWithItems(items, gen);
         }
 
+        writeLoop(o.loop(), gen);
+
         if (o.retry() != null) {
             gen.writeObjectField("retry", o.retry());
         }
diff --git a/runtime/v2/model/src/test/java/com/walmartlabs/concord/project/runtime/v2/ProjectSerializerV2Test.java b/runtime/v2/model/src/test/java/com/walmartlabs/concord/project/runtime/v2/ProjectSerializerV2Test.java
index 377222c295..ddb5d8a865 100644
--- a/runtime/v2/model/src/test/java/com/walmartlabs/concord/project/runtime/v2/ProjectSerializerV2Test.java
+++ b/runtime/v2/model/src/test/java/com/walmartlabs/concord/project/runtime/v2/ProjectSerializerV2Test.java
@@ -85,6 +85,7 @@ public void testFlowCall() throws Exception {
                 .putInput("in-1", "v1")
                 .addOut("o1")
                 .withItems(withItems())
+                .loop(serialLoop())
                 .retry(retry())
                 .errorSteps(steps())
                 .build();
@@ -197,6 +198,7 @@ public void testScriptCall() throws Exception {
                 .body("print(\"Hello, \", myVar)")
                 .putInput("in", "v")
                 .withItems(withItems())
+                .loop(serialLoop())
                 .retry(retry())
                 .errorSteps(steps())
                 .meta(meta())
@@ -233,6 +235,7 @@ public void testTaskCall() throws Exception {
                 .putInput("msg", "BOO")
                 .out("out")
                 .withItems(withItems())
+                .loop(serialLoop())
                 .retry(retry())
                 .errorSteps(steps())
                 .meta(meta())
@@ -313,7 +316,12 @@ public void testProcessDefinition() throws Exception {
                 .dest("dest")
                 .build()));
 
+        ProcessDefinitionConfiguration cfg = ProcessDefinitionConfiguration.builder()
+                .parallelLoopParallelism(123)
+                .build();
+
         ProcessDefinition pd = ProcessDefinition.builder()
+                .configuration(cfg)
                 .forms(forms)
                 .putFlows("flow1", steps())
                 .addPublicFlows("flow1")
@@ -360,6 +368,16 @@ private static WithItems parallelWithItems() {
         return WithItems.of(items, WithItems.Mode.PARALLEL);
     }
 
+    private static Loop serialLoop() {
+        ArrayList<String> items = new ArrayList<>();
+        items.add("item1");
+        items.add("item2");
+        return Loop.builder()
+                .items(items)
+                .mode(Loop.Mode.SERIAL)
+                .build();
+    }
+
     private static Retry retry() {
         return Retry.builder()
                 .times(1)
diff --git a/runtime/v2/model/src/test/java/com/walmartlabs/concord/project/runtime/v2/parser/YamlErrorParserTest.java b/runtime/v2/model/src/test/java/com/walmartlabs/concord/project/runtime/v2/parser/YamlErrorParserTest.java
index b4f1e7a42e..6fe8cfc500 100644
--- a/runtime/v2/model/src/test/java/com/walmartlabs/concord/project/runtime/v2/parser/YamlErrorParserTest.java
+++ b/runtime/v2/model/src/test/java/com/walmartlabs/concord/project/runtime/v2/parser/YamlErrorParserTest.java
@@ -796,7 +796,7 @@ public void test214() throws Exception {
     @Test
     public void test215() throws Exception {
         String msg =
-                "(015.yml): Error @ line: 15, col: 14. Unknown options: ['trash' [STRING] @ line: 15, col: 14], expected: [error, ignoreErrors, in, meta, name, out, parallelWithItems, retry, withItems]. Remove invalid options and/or fix indentation\n" +
+                "(015.yml): Error @ line: 15, col: 14. Unknown options: ['trash' [STRING] @ line: 15, col: 14], expected: [error, ignoreErrors, in, loop, meta, name, out, parallelWithItems, retry, withItems]. Remove invalid options and/or fix indentation\n" +
                         "\twhile processing steps:\n" +
                         "\t'task' @ line: 3, col: 7\n" +
                         "\t\t'main' @ line: 2, col: 3\n" +
@@ -857,6 +857,34 @@ public void test219() throws Exception {
         assertErrorMessage("errors/tasks/019.yml", msg);
     }
 
+    @Test
+    public void test220() throws Exception {
+        String msg =
+                "(020.yml): Error @ line: 7, col: 15. Invalid value: trash, expected: [SERIAL, PARALLEL]\n" +
+                        "\twhile processing steps:\n" +
+                        "\t'mode' @ line: 7, col: 9\n" +
+                        "\t\t'loop' @ line: 5, col: 7\n" +
+                        "\t\t\t'task' @ line: 4, col: 7\n" +
+                        "\t\t\t\t'main' @ line: 2, col: 3\n" +
+                        "\t\t\t\t\t'flows' @ line: 1, col: 1";
+
+        assertErrorMessage("errors/tasks/020.yml", msg);
+    }
+
+    @Test
+    public void test221() throws Exception {
+        String msg =
+                "(021.yml): Error @ line: 7, col: 22. Invalid value type, expected: INT, got: STRING\n" +
+                        "\twhile processing steps:\n" +
+                        "\t'parallelism' @ line: 7, col: 9\n" +
+                        "\t\t'loop' @ line: 5, col: 7\n" +
+                        "\t\t\t'task' @ line: 4, col: 7\n" +
+                        "\t\t\t\t'main' @ line: 2, col: 3\n" +
+                        "\t\t\t\t\t'flows' @ line: 1, col: 1";
+
+        assertErrorMessage("errors/tasks/021.yml", msg);
+    }
+
     @Test
     public void test300() throws Exception {
         String msg =
@@ -1043,7 +1071,7 @@ public void test314() throws Exception {
     @Test
     public void test315() throws Exception {
         String msg =
-                "(015.yml): Error @ line: 15, col: 14. Unknown options: ['trash' [STRING] @ line: 15, col: 14], expected: [error, in, meta, name, out, parallelWithItems, retry, withItems]. Remove invalid options and/or fix indentation\n" +
+                "(015.yml): Error @ line: 15, col: 14. Unknown options: ['trash' [STRING] @ line: 15, col: 14], expected: [error, in, loop, meta, name, out, parallelWithItems, retry, withItems]. Remove invalid options and/or fix indentation\n" +
                         "\twhile processing steps:\n" +
                         "\t'call' @ line: 3, col: 7\n" +
                         "\t\t'main' @ line: 2, col: 3\n" +
@@ -1332,7 +1360,7 @@ public void test702() throws Exception {
     @Test
     public void test703() throws Exception {
         String msg =
-                "(003.yml): Error @ line: 5, col: 13. Unknown options: ['trash' [NULL] @ line: 5, col: 13], expected: [error, meta, name, out, parallelWithItems, withItems]. Remove invalid options and/or fix indentation\n" +
+                "(003.yml): Error @ line: 5, col: 13. Unknown options: ['trash' [NULL] @ line: 5, col: 13], expected: [error, loop, meta, name, out, parallelWithItems, withItems]. Remove invalid options and/or fix indentation\n" +
                         "\twhile processing steps:\n" +
                         "\t'try' @ line: 3, col: 7\n" +
                         "\t\t'main' @ line: 2, col: 3\n" +
@@ -1383,7 +1411,7 @@ public void test706() throws Exception {
     @Test
     public void test707() throws Exception {
         String msg =
-                "(007.yml): Error @ line: 11, col: 13. Unknown options: ['trash' [NULL] @ line: 11, col: 13], expected: [error, meta, name, out, parallelWithItems, withItems]. Remove invalid options and/or fix indentation\n" +
+                "(007.yml): Error @ line: 11, col: 13. Unknown options: ['trash' [NULL] @ line: 11, col: 13], expected: [error, loop, meta, name, out, parallelWithItems, withItems]. Remove invalid options and/or fix indentation\n" +
                         "\twhile processing steps:\n" +
                         "\t'try' @ line: 3, col: 7\n" +
                         "\t\t'main' @ line: 2, col: 3\n" +
@@ -1881,7 +1909,7 @@ public void test1305_1() throws Exception {
     @Test
     public void test1306() throws Exception {
         String msg =
-                "(006.yml): Error @ line: 8, col: 9. Unknown options: ['trash' [NULL] @ line: 8, col: 9], expected: [activeProfiles, arguments, debug, dependencies, entryPoint, events, exclusive, meta, out, processTimeout, requirements, runtime, suspendTimeout, template]. Remove invalid options and/or fix indentation\n" +
+                "(006.yml): Error @ line: 8, col: 9. Unknown options: ['trash' [NULL] @ line: 8, col: 9], expected: [activeProfiles, arguments, debug, dependencies, entryPoint, events, exclusive, meta, out, parallelLoopParallelism, processTimeout, requirements, runtime, suspendTimeout, template]. Remove invalid options and/or fix indentation\n" +
                         "\twhile processing steps:\n" +
                         "\t'configuration' @ line: 1, col: 1";
 
@@ -2257,7 +2285,7 @@ public void test1701() throws Exception {
 
     @Test
     public void test1702() throws Exception {
-        String msg = "(002.yml): Error @ line: 4, col: 14. Unknown options: ['body1' [STRING] @ line: 4, col: 14], expected: [body, error, in, meta, name, out, parallelWithItems, retry, withItems]. Remove invalid options and/or fix indentation\n" +
+        String msg = "(002.yml): Error @ line: 4, col: 14. Unknown options: ['body1' [STRING] @ line: 4, col: 14], expected: [body, error, in, loop, meta, name, out, parallelWithItems, retry, withItems]. Remove invalid options and/or fix indentation\n" +
                 "\twhile processing steps:\n" +
                 "\t'script' @ line: 3, col: 7\n" +
                 "\t\t'main' @ line: 2, col: 3\n" +
diff --git a/runtime/v2/model/src/test/java/com/walmartlabs/concord/project/runtime/v2/parser/YamlOkParserTest.java b/runtime/v2/model/src/test/java/com/walmartlabs/concord/project/runtime/v2/parser/YamlOkParserTest.java
index ce5da37d8f..c1d443ff5a 100644
--- a/runtime/v2/model/src/test/java/com/walmartlabs/concord/project/runtime/v2/parser/YamlOkParserTest.java
+++ b/runtime/v2/model/src/test/java/com/walmartlabs/concord/project/runtime/v2/parser/YamlOkParserTest.java
@@ -56,6 +56,10 @@ public void test000() throws Exception {
         // withItems
         assertEquals(1, t.getOptions().withItems().value());
 
+        // loop
+        assertEquals(1, t.getOptions().loop().items());
+        assertEquals(Loop.Mode.SERIAL, t.getOptions().loop().mode());
+
         // input
         Map<String, Object> input = new HashMap<>();
         input.put("k", "v");
diff --git a/runtime/v2/model/src/test/resources/000.yml b/runtime/v2/model/src/test/resources/000.yml
index 31bcdd355a..7b92396c2c 100644
--- a/runtime/v2/model/src/test/resources/000.yml
+++ b/runtime/v2/model/src/test/resources/000.yml
@@ -8,6 +8,8 @@ flows:
         k2: 2
         k3: false
       withItems: 1
+      loop:
+        items: 1
       retry:
         times: 1
         delay: 2
diff --git a/runtime/v2/model/src/test/resources/errors/tasks/020.yml b/runtime/v2/model/src/test/resources/errors/tasks/020.yml
new file mode 100644
index 0000000000..1a31885578
--- /dev/null
+++ b/runtime/v2/model/src/test/resources/errors/tasks/020.yml
@@ -0,0 +1,7 @@
+flows:
+  main:
+    - name: "123123"
+      task: "boo"
+      loop:
+        items: 1
+        mode: trash
\ No newline at end of file
diff --git a/runtime/v2/model/src/test/resources/errors/tasks/021.yml b/runtime/v2/model/src/test/resources/errors/tasks/021.yml
new file mode 100644
index 0000000000..582f8dc1d8
--- /dev/null
+++ b/runtime/v2/model/src/test/resources/errors/tasks/021.yml
@@ -0,0 +1,7 @@
+flows:
+  main:
+    - name: "123123"
+      task: "boo"
+      loop:
+        items: 1
+        parallelism: one
\ No newline at end of file
diff --git a/runtime/v2/model/src/test/resources/serializer/flowCallStep.yml b/runtime/v2/model/src/test/resources/serializer/flowCallStep.yml
index 0f8e45ff11..788f63978e 100644
--- a/runtime/v2/model/src/test/resources/serializer/flowCallStep.yml
+++ b/runtime/v2/model/src/test/resources/serializer/flowCallStep.yml
@@ -6,6 +6,11 @@ out:
 withItems:
 - "item1"
 - "item2"
+loop:
+  mode: "serial"
+  items:
+  - "item1"
+  - "item2"
 retry:
   times: "${times}"
   delay: "${delay}"
diff --git a/runtime/v2/model/src/test/resources/serializer/processDefinition.yml b/runtime/v2/model/src/test/resources/serializer/processDefinition.yml
index b21afb00eb..00c35869b9 100644
--- a/runtime/v2/model/src/test/resources/serializer/processDefinition.yml
+++ b/runtime/v2/model/src/test/resources/serializer/processDefinition.yml
@@ -19,6 +19,7 @@ configuration:
     - "vaultPassword"
     recordTaskMeta: false
     truncateMeta: true
+  parallelLoopParallelism: 123
 flows:
   flow1:
   - checkpoint: "chp1"
diff --git a/runtime/v2/model/src/test/resources/serializer/scriptStep.yml b/runtime/v2/model/src/test/resources/serializer/scriptStep.yml
index e8ec499081..664d3b1661 100644
--- a/runtime/v2/model/src/test/resources/serializer/scriptStep.yml
+++ b/runtime/v2/model/src/test/resources/serializer/scriptStep.yml
@@ -5,6 +5,11 @@ in:
 withItems:
 - "item1"
 - "item2"
+loop:
+  mode: "serial"
+  items:
+  - "item1"
+  - "item2"
 retry:
   times: "${times}"
   delay: "${delay}"
diff --git a/runtime/v2/model/src/test/resources/serializer/taskCallStep.yml b/runtime/v2/model/src/test/resources/serializer/taskCallStep.yml
index 175a1b439c..6e1e1a6019 100644
--- a/runtime/v2/model/src/test/resources/serializer/taskCallStep.yml
+++ b/runtime/v2/model/src/test/resources/serializer/taskCallStep.yml
@@ -3,6 +3,11 @@ out: "out"
 withItems:
 - "item1"
 - "item2"
+loop:
+  mode: "serial"
+  items:
+  - "item1"
+  - "item2"
 retry:
   times: "${times}"
   delay: "${delay}"
diff --git a/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/compiler/FlowCallCompiler.java b/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/compiler/FlowCallCompiler.java
index 4a1fe3a260..a53aa59617 100644
--- a/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/compiler/FlowCallCompiler.java
+++ b/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/compiler/FlowCallCompiler.java
@@ -21,10 +21,7 @@
  */
 
 import com.walmartlabs.concord.runtime.v2.model.*;
-import com.walmartlabs.concord.runtime.v2.runner.vm.ErrorWrapper;
-import com.walmartlabs.concord.runtime.v2.runner.vm.FlowCallCommand;
-import com.walmartlabs.concord.runtime.v2.runner.vm.RetryWrapper;
-import com.walmartlabs.concord.runtime.v2.runner.vm.WithItemsWrapper;
+import com.walmartlabs.concord.runtime.v2.runner.vm.*;
 import com.walmartlabs.concord.svm.Command;
 
 import javax.inject.Named;
@@ -55,6 +52,11 @@ public Command compile(CompilerContext context, FlowCall step) {
             cmd = WithItemsWrapper.of(cmd, withItems, options.out(), options.outExpr());
         }
 
+        Loop loop = options.loop();
+        if (loop != null) {
+            cmd = LoopWrapper.of(context, cmd, loop, options.out(), options.outExpr());
+        }
+
         List<Step> errorSteps = options.errorSteps();
         if (!errorSteps.isEmpty()) {
             cmd = new ErrorWrapper(cmd, CompilerUtils.compile(context, errorSteps));
diff --git a/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/compiler/GroupOfStepsCompiler.java b/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/compiler/GroupOfStepsCompiler.java
index 6270ec3747..5e894b1625 100644
--- a/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/compiler/GroupOfStepsCompiler.java
+++ b/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/compiler/GroupOfStepsCompiler.java
@@ -20,12 +20,10 @@
  * =====
  */
 
-import com.walmartlabs.concord.runtime.v2.model.GroupOfSteps;
-import com.walmartlabs.concord.runtime.v2.model.GroupOfStepsOptions;
-import com.walmartlabs.concord.runtime.v2.model.Step;
-import com.walmartlabs.concord.runtime.v2.model.WithItems;
+import com.walmartlabs.concord.runtime.v2.model.*;
 import com.walmartlabs.concord.runtime.v2.runner.vm.BlockCommand;
 import com.walmartlabs.concord.runtime.v2.runner.vm.ErrorWrapper;
+import com.walmartlabs.concord.runtime.v2.runner.vm.LoopWrapper;
 import com.walmartlabs.concord.runtime.v2.runner.vm.WithItemsWrapper;
 import com.walmartlabs.concord.svm.Command;
 
@@ -53,6 +51,11 @@ public Command compile(CompilerContext context, GroupOfSteps step) {
             return WithItemsWrapper.of(cmd, withItems, options.out(), Collections.emptyMap());
         }
 
+        Loop loop = options.loop();
+        if (loop != null) {
+            cmd = LoopWrapper.of(context, cmd, loop, options.out(), Collections.emptyMap());
+        }
+
         List<Step> errorSteps = options.errorSteps();
         if (!options.errorSteps().isEmpty()) {
             cmd = new ErrorWrapper(cmd, compile(context, errorSteps));
diff --git a/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/compiler/ScriptCallCompiler.java b/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/compiler/ScriptCallCompiler.java
index 27a42a9c52..8285d03306 100644
--- a/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/compiler/ScriptCallCompiler.java
+++ b/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/compiler/ScriptCallCompiler.java
@@ -21,10 +21,7 @@
  */
 
 import com.walmartlabs.concord.runtime.v2.model.*;
-import com.walmartlabs.concord.runtime.v2.runner.vm.ErrorWrapper;
-import com.walmartlabs.concord.runtime.v2.runner.vm.RetryWrapper;
-import com.walmartlabs.concord.runtime.v2.runner.vm.ScriptCallCommand;
-import com.walmartlabs.concord.runtime.v2.runner.vm.WithItemsWrapper;
+import com.walmartlabs.concord.runtime.v2.runner.vm.*;
 import com.walmartlabs.concord.svm.Command;
 
 import javax.inject.Named;
@@ -56,6 +53,11 @@ public Command compile(CompilerContext context, ScriptCall step) {
             cmd = WithItemsWrapper.of(cmd, withItems, Collections.emptyList(), Collections.emptyMap());
         }
 
+        Loop loop = options.loop();
+        if (loop != null) {
+            cmd = LoopWrapper.of(context, cmd, loop, Collections.emptyList(), Collections.emptyMap());
+        }
+
         List<Step> errorSteps = options.errorSteps();
         if (!errorSteps.isEmpty()) {
             cmd = new ErrorWrapper(cmd, CompilerUtils.compile(context, errorSteps));
diff --git a/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/compiler/TaskCallCompiler.java b/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/compiler/TaskCallCompiler.java
index a893c1cfce..f81d949463 100644
--- a/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/compiler/TaskCallCompiler.java
+++ b/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/compiler/TaskCallCompiler.java
@@ -21,10 +21,7 @@
  */
 
 import com.walmartlabs.concord.runtime.v2.model.*;
-import com.walmartlabs.concord.runtime.v2.runner.vm.ErrorWrapper;
-import com.walmartlabs.concord.runtime.v2.runner.vm.RetryWrapper;
-import com.walmartlabs.concord.runtime.v2.runner.vm.TaskCallCommand;
-import com.walmartlabs.concord.runtime.v2.runner.vm.WithItemsWrapper;
+import com.walmartlabs.concord.runtime.v2.runner.vm.*;
 import com.walmartlabs.concord.svm.Command;
 
 import javax.inject.Named;
@@ -61,6 +58,15 @@ public Command compile(CompilerContext context, TaskCall step) {
             cmd = WithItemsWrapper.of(cmd, withItems, out, options.outExpr());
         }
 
+        Loop loop = options.loop();
+        if (loop != null) {
+            Collection<String> out = Collections.emptyList();
+            if (options.out() != null) {
+                out = Collections.singletonList(options.out());
+            }
+            cmd = LoopWrapper.of(context, cmd, loop, out, options.outExpr());
+        }
+
         List<Step> errorSteps = options.errorSteps();
         if (!errorSteps.isEmpty()) {
             cmd = new ErrorWrapper(cmd, CompilerUtils.compile(context, errorSteps));
diff --git a/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/el/LazyExpressionEvaluator.java b/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/el/LazyExpressionEvaluator.java
index c69130406b..0db24c1921 100644
--- a/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/el/LazyExpressionEvaluator.java
+++ b/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/el/LazyExpressionEvaluator.java
@@ -23,10 +23,7 @@
 import com.walmartlabs.concord.common.ConfigurationUtils;
 import com.walmartlabs.concord.runtime.v2.runner.el.functions.*;
 import com.walmartlabs.concord.runtime.v2.runner.el.resolvers.BeanELResolver;
-import com.walmartlabs.concord.runtime.v2.runner.el.resolvers.MethodAccessorResolver;
-import com.walmartlabs.concord.runtime.v2.runner.el.resolvers.TaskMethodResolver;
-import com.walmartlabs.concord.runtime.v2.runner.el.resolvers.TaskResolver;
-import com.walmartlabs.concord.runtime.v2.runner.el.resolvers.VariableResolver;
+import com.walmartlabs.concord.runtime.v2.runner.el.resolvers.*;
 import com.walmartlabs.concord.runtime.v2.runner.tasks.TaskProviders;
 
 import javax.el.*;
diff --git a/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/el/functions/CurrentFlowNameFunction.java b/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/el/functions/CurrentFlowNameFunction.java
index 25b069532f..bf3003d935 100644
--- a/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/el/functions/CurrentFlowNameFunction.java
+++ b/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/el/functions/CurrentFlowNameFunction.java
@@ -20,10 +20,8 @@
  * =====
  */
 
-import com.walmartlabs.concord.runtime.v2.ProcessDefinitionUtils;
 import com.walmartlabs.concord.runtime.v2.runner.el.ThreadLocalEvalContext;
 import com.walmartlabs.concord.runtime.v2.sdk.Context;
-import com.walmartlabs.concord.runtime.v2.sdk.Execution;
 
 import java.lang.reflect.Method;
 
diff --git a/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/vm/LoopWrapper.java b/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/vm/LoopWrapper.java
new file mode 100644
index 0000000000..702b0fc230
--- /dev/null
+++ b/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/vm/LoopWrapper.java
@@ -0,0 +1,379 @@
+package com.walmartlabs.concord.runtime.v2.runner.vm;
+
+/*-
+ * *****
+ * Concord
+ * -----
+ * Copyright (C) 2017 - 2019 Walmart Inc.
+ * -----
+ * 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.
+ * =====
+ */
+
+import com.walmartlabs.concord.runtime.v2.model.Loop;
+import com.walmartlabs.concord.runtime.v2.model.Step;
+import com.walmartlabs.concord.runtime.v2.runner.compiler.CompilerContext;
+import com.walmartlabs.concord.runtime.v2.runner.context.ContextFactory;
+import com.walmartlabs.concord.runtime.v2.runner.el.EvalContextFactory;
+import com.walmartlabs.concord.runtime.v2.runner.el.ExpressionEvaluator;
+import com.walmartlabs.concord.runtime.v2.sdk.Context;
+import com.walmartlabs.concord.sdk.MapUtils;
+import com.walmartlabs.concord.svm.Runtime;
+import com.walmartlabs.concord.svm.*;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+import java.util.*;
+import java.util.stream.Collectors;
+
+public abstract class LoopWrapper implements Command {
+
+    public static LoopWrapper of(CompilerContext ctx, Command cmd, Loop withItems, Collection<String> outVariables, Map<String, Serializable> outExpressions) {
+        Collection<String> out = Collections.emptyList();
+        if (!outExpressions.isEmpty()) {
+            out = outExpressions.keySet();
+        } else if (!outVariables.isEmpty()) {
+            out = outVariables;
+        }
+
+        Loop.Mode mode = withItems.mode();
+        switch (mode) {
+            case SERIAL:
+                return new SerialWithItems(cmd, withItems, out);
+            case PARALLEL:
+                return new ParallelWithItems(ctx, cmd, withItems, out);
+            default:
+                throw new IllegalArgumentException("Unknown withItems mode: " + mode);
+        }
+    }
+
+    private static final long serialVersionUID = 1L;
+
+    // TODO move into the actual Constants
+    public static final String CURRENT_ITEMS = "items";
+    public static final String CURRENT_INDEX = "itemIndex";
+    public static final String CURRENT_ITEM = "item";
+
+    protected final Command cmd;
+    protected final Serializable items;
+    protected final Collection<String> outVariables;
+
+    protected LoopWrapper(Command cmd, Serializable items, Collection<String> outVariables) {
+        this.cmd = cmd;
+        this.items = items;
+        this.outVariables = outVariables;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public void eval(Runtime runtime, State state, ThreadId threadId) {
+        Frame frame = state.peekFrame(threadId);
+        frame.pop();
+
+        Serializable value = items;
+        if (value == null) {
+            // value is null, not going to run the wrapped command at all
+            return;
+        }
+
+        Step currentStep = null;
+        if (cmd instanceof StepCommand) {
+            currentStep = ((StepCommand<?>) cmd).getStep();
+        }
+
+        // create the context explicitly
+        ContextFactory contextFactory = runtime.getService(ContextFactory.class);
+        Context ctx = contextFactory.create(runtime, state, threadId, currentStep);
+
+        ExpressionEvaluator ee = runtime.getService(ExpressionEvaluator.class);
+        value = ee.eval(EvalContextFactory.global(ctx), value, Serializable.class);
+
+        // prepare items
+        // store items in an ArrayList because it is Serializable
+        ArrayList<Serializable> items;
+        if (value == null) {
+            // value is null, not going to run the wrapped command at all
+            return;
+        } else if (value instanceof Collection) {
+            Collection<Serializable> v = (Collection<Serializable>) value;
+            if (v.isEmpty()) {
+                // no items, nothing to do
+                return;
+            }
+
+            items = new ArrayList<>(v);
+        } else if (value instanceof Map) {
+            Map<Serializable, Serializable> m = (Map<Serializable, Serializable>) value;
+            items = m.entrySet().stream()
+                    .map(e -> new AbstractMap.SimpleImmutableEntry<>(e.getKey(), e.getValue()))
+                    .collect(Collectors.toCollection(ArrayList::new));
+        } else if (value.getClass().isArray()) {
+            items = new ArrayList<>(Arrays.asList((Serializable[]) value));
+        } else {
+            throw new IllegalArgumentException("'withItems' accepts only Lists of items, Java Maps or arrays of values. Got: " + value.getClass());
+        }
+
+        items.forEach(LoopWrapper::assertItem);
+
+        if (items.isEmpty()) {
+            return;
+        }
+
+        eval(state, threadId, items);
+    }
+
+    protected abstract void eval(State state, ThreadId threadId, ArrayList<Serializable> items);
+
+    static void assertItem(Object item) {
+        if (item == null) {
+            return;
+        }
+
+        try (ObjectOutputStream oos = new ObjectOutputStream(new ByteArrayOutputStream())) {
+            oos.writeObject(item);
+        } catch (IOException e) {
+            throw new IllegalArgumentException("Can't use non-serializable values in 'withItems': " + item + " (" + item.getClass() + ")");
+        }
+    }
+
+    static class ParallelWithItems extends LoopWrapper {
+
+        private static final long serialVersionUID = 1L;
+
+        private final int batchSize;
+
+        protected ParallelWithItems(CompilerContext ctx, Command cmd, Loop loop, Collection<String> outVariables) {
+            super(cmd, loop.items(), outVariables);
+
+            this.batchSize = batchSize(ctx, loop);
+        }
+
+        @Override
+        protected void eval(State state, ThreadId threadId, ArrayList<Serializable> items) {
+            // target frame for out variables
+            Frame targetFrame = VMUtils.assertNearestRoot(state, threadId);
+
+            List<ArrayList<Serializable>> batches = batches(items, batchSize);
+            for (ArrayList<Serializable> batch : batches) {
+                evalBatch(state, threadId, targetFrame, batch);
+            }
+
+            state.pushFrame(threadId, Frame.builder()
+                    .commands(new PrepareOutVariables(outVariables, targetFrame))
+                    .nonRoot()
+                    .build());
+        }
+
+        private void evalBatch(State state, ThreadId threadId, Frame targetFrame, ArrayList<Serializable> items) {
+            Frame frame = state.peekFrame(threadId);
+
+            List<Map.Entry<ThreadId, Serializable>> forks = items.stream()
+                    .map(e -> new AbstractMap.SimpleEntry<>(state.nextThreadId(), e))
+                    .collect(Collectors.toList());
+
+            for (int i = 0; i < forks.size(); i++) {
+                Map.Entry<ThreadId, Serializable> f = forks.get(i);
+
+                Frame cmdFrame = Frame.builder()
+                        .nonRoot()
+                        .build();
+
+                cmdFrame.setLocal(CURRENT_ITEMS, items);
+                cmdFrame.setLocal(CURRENT_INDEX, i);
+                cmdFrame.setLocal(CURRENT_ITEM, f.getValue());
+
+                // fork will create rootFrame for forked commands
+                Command itemCmd = new ForkCommand(f.getKey(),
+                        new AppendVariablesCommand(outVariables, null, targetFrame),
+                        cmd);
+                cmdFrame.push(itemCmd);
+
+                state.pushFrame(threadId, cmdFrame);
+            }
+
+            frame.push(new JoinCommand(forks.stream().map(Map.Entry::getKey).collect(Collectors.toSet())));
+        }
+
+        private static int batchSize(CompilerContext ctx, Loop loop) {
+            int result = MapUtils.getInt(loop.options(), "parallelism", -1);
+            if (result > 0) {
+                return result;
+            }
+            return ctx.processDefinition().configuration().parallelLoopParallelism();
+        }
+
+        private static List<ArrayList<Serializable>> batches(ArrayList<Serializable> items, int batchSize) {
+            List<ArrayList<Serializable>> result = new ArrayList<>();
+            for (int i = 0; i < items.size(); i += batchSize) {
+                result.add(new ArrayList<>(items.subList(i, Math.min(items.size(), i + batchSize))));
+            }
+            return result;
+        }
+    }
+
+    /**
+     * Wraps a command into a loop specified by {@code withItems} option.
+     * Creates a new call frame and keeps the item list, the current item
+     * and the index as frame-local variables.
+     */
+    static class SerialWithItems extends LoopWrapper {
+
+        private static final long serialVersionUID = 1L;
+
+        protected SerialWithItems(Command cmd, Loop loop, Collection<String> outVariables) {
+            super(cmd, loop.items(), outVariables);
+        }
+
+        @Override
+        protected void eval(State state, ThreadId threadId, ArrayList<Serializable> items) {
+            Frame loop = Frame.builder()
+                    .nonRoot()
+                    .build();
+
+            loop.setLocal(CURRENT_ITEMS, items);
+            loop.setLocal(CURRENT_INDEX, 0);
+            loop.setLocal(CURRENT_ITEM, items.get(0));
+
+            loop.push(new WithItemsNext(outVariables, cmd)); // next iteration
+
+            Frame cmdFrame = Frame.builder()
+                    .commands(cmd)
+                    .root()
+                    .build();
+
+            Frame targetFrame = VMUtils.assertNearestRoot(state, threadId);
+            loop.push(new AppendVariablesCommand(outVariables, cmdFrame, targetFrame));
+            loop.push(new PrepareOutVariables(outVariables, targetFrame));
+
+            state.pushFrame(threadId, loop);
+            state.pushFrame(threadId, cmdFrame);
+        }
+    }
+
+    static class WithItemsNext implements Command {
+
+        private static final long serialVersionUID = 1L;
+
+        private final Collection<String> outVariables;
+        private final Command cmd;
+
+        public WithItemsNext(Collection<String> outVariables, Command cmd) {
+            this.outVariables = outVariables;
+            this.cmd = cmd;
+        }
+
+        @Override
+        public void eval(Runtime runtime, State state, ThreadId threadId) {
+            Frame loop = state.peekFrame(threadId);
+            loop.pop();
+
+            List<Serializable> items = VMUtils.assertLocal(state, threadId, CURRENT_ITEMS);
+
+            int index = VMUtils.assertLocal(state, threadId, CURRENT_INDEX);
+            if (index + 1 >= items.size()) {
+                // end of the line, do nothing
+                return;
+            }
+
+            int newIndex = index + 1;
+            loop.setLocal(CURRENT_INDEX, newIndex);
+            loop.setLocal(CURRENT_ITEM, items.get(newIndex));
+
+            loop.push(new WithItemsNext(outVariables, cmd)); // next iteration
+
+            // frame wrapped command
+            Frame cmdFrame = Frame.builder()
+                    .commands(cmd)
+                    .root()
+                    .build();
+
+            Frame targetFrame = VMUtils.assertNearestRoot(state, threadId);
+            loop.push(new AppendVariablesCommand(outVariables, cmdFrame, targetFrame));
+
+            state.pushFrame(threadId, cmdFrame);
+        }
+    }
+
+    static class PrepareOutVariables implements Command {
+
+        private static final long serialVersionUID = 1L;
+
+        private final Collection<String> outVars;
+        private final Frame targetFrame;
+
+        private PrepareOutVariables(Collection<String> outVars, Frame targetFrame) {
+            this.outVars = outVars;
+            this.targetFrame = targetFrame;
+        }
+
+        @Override
+        public void eval(Runtime runtime, State state, ThreadId threadId) {
+            Frame frame = state.peekFrame(threadId);
+            frame.pop();
+
+            if (outVars.isEmpty()) {
+                return;
+            }
+
+            for (String outVar : outVars) {
+                VMUtils.putLocal(targetFrame, outVar, new ArrayList<>());
+            }
+        }
+    }
+
+    /**
+     * Appends values of the specified variables from the source frame into
+     * list variables in the target frame.
+     */
+    static class AppendVariablesCommand implements Command {
+
+        private static final long serialVersionUID = 1L;
+
+        private final Collection<String> variables;
+        private final Frame sourceFrame;
+        private final Frame targetFrame;
+
+        public AppendVariablesCommand(Collection<String> variables, Frame sourceFrame, Frame targetFrame) {
+            this.variables = variables;
+            this.sourceFrame = sourceFrame;
+            this.targetFrame = targetFrame;
+        }
+
+        @Override
+        @SuppressWarnings("unchecked")
+        public void eval(Runtime runtime, State state, ThreadId threadId) {
+            Frame frame = state.peekFrame(threadId);
+            frame.pop();
+
+            if (variables.isEmpty()) {
+                return;
+            }
+
+            Frame effectiveSourceFrame = sourceFrame != null ? sourceFrame : VMUtils.assertNearestRoot(state, threadId);
+
+            for (String v : variables) {
+                // make sure we're not modifying the same list concurrently
+                synchronized (targetFrame) {
+                    ArrayList<Serializable> results = (ArrayList<Serializable>) targetFrame.getLocal(v);
+                    Serializable result = null;
+                    if (effectiveSourceFrame.hasLocal(v)) {
+                        result = effectiveSourceFrame.getLocal(v);
+                    }
+                    results.add(result);
+                }
+            }
+        }
+    }
+}
diff --git a/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/vm/ScriptCallCommand.java b/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/vm/ScriptCallCommand.java
index b399c1ccf7..a15a577d40 100644
--- a/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/vm/ScriptCallCommand.java
+++ b/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/vm/ScriptCallCommand.java
@@ -28,7 +28,6 @@
 import com.walmartlabs.concord.runtime.v2.runner.script.ScriptEvaluator;
 import com.walmartlabs.concord.runtime.v2.runner.script.ScriptResult;
 import com.walmartlabs.concord.runtime.v2.sdk.Context;
-import com.walmartlabs.concord.runtime.v2.sdk.TaskResult;
 import com.walmartlabs.concord.svm.Runtime;
 import com.walmartlabs.concord.svm.State;
 import com.walmartlabs.concord.svm.ThreadId;
@@ -36,7 +35,6 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.*;
-import java.util.Collections;
 import java.util.Map;
 import java.util.Objects;
 
diff --git a/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/vm/StepExecutionException.java b/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/vm/StepExecutionException.java
new file mode 100644
index 0000000000..15b6f82145
--- /dev/null
+++ b/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/vm/StepExecutionException.java
@@ -0,0 +1,46 @@
+package com.walmartlabs.concord.runtime.v2.runner.vm;
+
+/*-
+ * *****
+ * Concord
+ * -----
+ * Copyright (C) 2017 - 2022 Walmart Inc.
+ * -----
+ * 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.
+ * =====
+ */
+
+import com.walmartlabs.concord.runtime.v2.model.Location;
+import com.walmartlabs.concord.runtime.v2.model.Step;
+
+public class StepExecutionException extends RuntimeException {
+
+    private static final long serialVersionUID = 1L;
+
+    private final Step step;
+
+    public StepExecutionException(Step step, Exception cause) {
+        super(cause);
+
+        this.step = step;
+    }
+
+    public Step getStep() {
+        return step;
+    }
+
+    @Override
+    public String getMessage() {
+        return Location.toErrorPrefix(step.getLocation()) + getCause().getMessage();
+    }
+}
diff --git a/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/vm/TaskCallUtils.java b/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/vm/TaskCallUtils.java
index 2801ca8a9e..83aeb4c3ea 100644
--- a/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/vm/TaskCallUtils.java
+++ b/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/vm/TaskCallUtils.java
@@ -21,16 +21,12 @@
  */
 
 import com.walmartlabs.concord.runtime.v2.model.TaskCallOptions;
-import com.walmartlabs.concord.runtime.v2.runner.el.EvalContextFactory;
-import com.walmartlabs.concord.runtime.v2.runner.el.ExpressionEvaluator;
 import com.walmartlabs.concord.runtime.v2.sdk.Context;
 import com.walmartlabs.concord.runtime.v2.sdk.TaskResult;
 import com.walmartlabs.concord.svm.Runtime;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.Serializable;
-import java.util.Collections;
 import java.util.Map;
 
 public final class TaskCallUtils {
diff --git a/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/vm/WithItemsWrapper.java b/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/vm/WithItemsWrapper.java
index ede7c9c739..bfe47c352f 100644
--- a/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/vm/WithItemsWrapper.java
+++ b/runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/vm/WithItemsWrapper.java
@@ -36,6 +36,10 @@
 import java.util.*;
 import java.util.stream.Collectors;
 
+/**
+ * @deprecated use {@link com.walmartlabs.concord.runtime.v2.runner.vm.LoopWrapper}
+ */
+@Deprecated
 public abstract class WithItemsWrapper implements Command {
 
     public static WithItemsWrapper of(Command cmd, WithItems withItems, Collection<String> outVariables, Map<String, Serializable> outExpressions) {
diff --git a/runtime/v2/runner/src/test/java/com/walmartlabs/concord/runtime/v2/runner/MainTest.java b/runtime/v2/runner/src/test/java/com/walmartlabs/concord/runtime/v2/runner/MainTest.java
index 67e4239c79..5d3ea2932f 100644
--- a/runtime/v2/runner/src/test/java/com/walmartlabs/concord/runtime/v2/runner/MainTest.java
+++ b/runtime/v2/runner/src/test/java/com/walmartlabs/concord/runtime/v2/runner/MainTest.java
@@ -556,6 +556,18 @@ public void testParallelWithItemsTask() throws Exception {
         assertLog(log, ".*threadIds: \\[1, 2, 3].*");
     }
 
+    @Test
+    public void testParallelLoopTask() throws Exception {
+        deploy("parallelLoopTask");
+
+        save(ProcessConfiguration.builder()
+                .build());
+
+        byte[] log = run();
+        assertLog(log, ".*result: \\[10, 20, 30\\].*");
+        assertLog(log, ".*threadIds: \\[1, 2, 3].*");
+    }
+
     @Test
     public void testWithItemsBlock() throws Exception {
         deploy("withItemsBlock");
@@ -567,6 +579,17 @@ public void testWithItemsBlock() throws Exception {
         assertLog(log, ".*result: \\[10, 20, 30\\].*");
     }
 
+    @Test
+    public void testLoopBlock() throws Exception {
+        deploy("loopBlock");
+
+        save(ProcessConfiguration.builder()
+                .build());
+
+        byte[] log = run();
+        assertLog(log, ".*result: \\[10, 20, 30\\].*");
+    }
+
     @Test
     public void testWithItemsSet() throws Exception {
         deploy("withItemsSet");
@@ -581,6 +604,20 @@ public void testWithItemsSet() throws Exception {
         assertLog(log, ".*after add: \\[3\\].*");
     }
 
+    @Test
+    public void testLoopSet() throws Exception {
+        deploy("loopSet");
+
+        save(ProcessConfiguration.builder()
+                .build());
+
+        byte[] log = run();
+        assertLogAtLeast(log, 3, ".*empty: \\[\\].*");
+        assertLog(log, ".*after add: \\[1\\].*");
+        assertLog(log, ".*after add: \\[2\\].*");
+        assertLog(log, ".*after add: \\[3\\].*");
+    }
+
     @Test
     public void testUnknownMethod() throws Exception {
         deploy("unknownMethod");
diff --git a/runtime/v2/runner/src/test/java/com/walmartlabs/concord/runtime/v2/runner/el/ImmutablesTest.java b/runtime/v2/runner/src/test/java/com/walmartlabs/concord/runtime/v2/runner/el/ImmutablesTest.java
index 430cf848e1..b41437a455 100644
--- a/runtime/v2/runner/src/test/java/com/walmartlabs/concord/runtime/v2/runner/el/ImmutablesTest.java
+++ b/runtime/v2/runner/src/test/java/com/walmartlabs/concord/runtime/v2/runner/el/ImmutablesTest.java
@@ -28,7 +28,7 @@
 import java.util.Collections;
 import java.util.Map;
 
-import static org.junit.jupiter.api.Assertions.*;
+import static org.junit.jupiter.api.Assertions.assertEquals;
 
 public class ImmutablesTest {
 
diff --git a/runtime/v2/runner/src/test/java/com/walmartlabs/concord/runtime/v2/runner/el/MethodNotFoundExceptionTest.java b/runtime/v2/runner/src/test/java/com/walmartlabs/concord/runtime/v2/runner/el/MethodNotFoundExceptionTest.java
index 0537835058..5f3ec8534a 100644
--- a/runtime/v2/runner/src/test/java/com/walmartlabs/concord/runtime/v2/runner/el/MethodNotFoundExceptionTest.java
+++ b/runtime/v2/runner/src/test/java/com/walmartlabs/concord/runtime/v2/runner/el/MethodNotFoundExceptionTest.java
@@ -24,7 +24,7 @@
 
 import java.util.List;
 
-import static org.junit.jupiter.api.Assertions.*;
+import static org.junit.jupiter.api.Assertions.assertEquals;
 
 public class MethodNotFoundExceptionTest {
 
diff --git a/runtime/v2/runner/src/test/resources/com/walmartlabs/concord/runtime/v2/runner/loopBlock/concord.yml b/runtime/v2/runner/src/test/resources/com/walmartlabs/concord/runtime/v2/runner/loopBlock/concord.yml
new file mode 100644
index 0000000000..6e6caf1fc6
--- /dev/null
+++ b/runtime/v2/runner/src/test/resources/com/walmartlabs/concord/runtime/v2/runner/loopBlock/concord.yml
@@ -0,0 +1,13 @@
+flows:
+  default:
+    - block:
+        - set:
+            x: "${item * 10}"
+      out: x
+      loop:
+        items:
+          - 1
+          - 2
+          - 3
+
+    - log: "result: ${x}"
diff --git a/runtime/v2/runner/src/test/resources/com/walmartlabs/concord/runtime/v2/runner/loopSet/concord.yml b/runtime/v2/runner/src/test/resources/com/walmartlabs/concord/runtime/v2/runner/loopSet/concord.yml
new file mode 100644
index 0000000000..71064d2747
--- /dev/null
+++ b/runtime/v2/runner/src/test/resources/com/walmartlabs/concord/runtime/v2/runner/loopSet/concord.yml
@@ -0,0 +1,15 @@
+flows:
+  default:
+    - set:
+        items: [1,2,3]
+
+    - call: test
+      loop:
+        items: ${items}
+
+  test:
+    - set:
+        arr: []
+    - log: "empty: ${arr}"
+    - ${arr.add(item)}
+    - log: "after add: ${arr}"
\ No newline at end of file
diff --git a/runtime/v2/runner/src/test/resources/com/walmartlabs/concord/runtime/v2/runner/parallelLoopTask/concord.yml b/runtime/v2/runner/src/test/resources/com/walmartlabs/concord/runtime/v2/runner/parallelLoopTask/concord.yml
new file mode 100644
index 0000000000..ff8f6cd64a
--- /dev/null
+++ b/runtime/v2/runner/src/test/resources/com/walmartlabs/concord/runtime/v2/runner/parallelLoopTask/concord.yml
@@ -0,0 +1,15 @@
+flows:
+  default:
+    - task: resultTask
+      in:
+        result: "${item * 10}"
+      out: x
+      loop:
+        mode: parallel
+        items:
+          - 1
+          - 2
+          - 3
+
+    - log: "result: ${x.stream().map(v -> v.result).sorted().toList()}"
+    - log: "threadIds: ${x.stream().map(v -> v.threadId).sorted().toList()}"
diff --git a/runtime/v2/runner/src/test/resources/com/walmartlabs/concord/runtime/v2/runner/rethrow/concord.yml b/runtime/v2/runner/src/test/resources/com/walmartlabs/concord/runtime/v2/runner/rethrow/concord.yml
new file mode 100644
index 0000000000..02919d5af4
--- /dev/null
+++ b/runtime/v2/runner/src/test/resources/com/walmartlabs/concord/runtime/v2/runner/rethrow/concord.yml
@@ -0,0 +1,5 @@
+flows:
+  default:
+    - log: "before exit"
+    - exit
+    - log: "after exit"