Added forkFn support and parallel

Signed-off-by: Ricardo Zanini <[email protected]>

Author: ricardozanini

fluent/agentic/src/main/java/io/serverlessworkflow/fluent/agentic/AgentDoTaskFluent.java

@@ -37,4 +37,10 @@ default SELF sequence(Object... agents) {
   default SELF loop(Consumer<LoopAgentsBuilder> builder) {
     return loop("loop-" + UUID.randomUUID(), builder);
   }
+
+  SELF parallel(String name, Object... agents);
+
+  default SELF parallel(Object... agents) {
+    return parallel("par-" + UUID.randomUUID(), agents);
+  }
 }

fluent/agentic/src/main/java/io/serverlessworkflow/fluent/agentic/AgentTaskItemListBuilder.java

@@ -15,10 +15,12 @@
  */
 package io.serverlessworkflow.fluent.agentic;
 
+import dev.langchain4j.agentic.internal.AgentExecutor;
 import io.serverlessworkflow.api.types.Task;
 import io.serverlessworkflow.api.types.TaskItem;
 import io.serverlessworkflow.fluent.func.FuncTaskItemListBuilder;
 import io.serverlessworkflow.fluent.spec.BaseTaskItemListBuilder;
+import java.util.List;
 import java.util.function.Consumer;
 
 public class AgentTaskItemListBuilder extends BaseTaskItemListBuilder<AgentTaskItemListBuilder>
@@ -42,7 +44,7 @@ public AgentTaskItemListBuilder agent(String name, Object agent) {
         .forEach(
             exec ->
                 this.funcDelegate.callFn(name, fn -> fn.function(AgentAdapters.toFunction(exec))));
-    return this;
+    return self();
   }
 
   @Override
@@ -61,6 +63,20 @@ public AgentTaskItemListBuilder loop(String name, Consumer<LoopAgentsBuilder> co
     return self();
   }
 
+  @Override
+  public AgentTaskItemListBuilder parallel(String name, Object... agents) {
+    this.funcDelegate.forkFn(
+        name,
+        fork -> {
+          List<AgentExecutor> execs = AgentAdapters.toExecutors(agents);
+          for (int i = 0; i < execs.size(); i++) {
+            AgentExecutor ex = execs.get(i);
+            fork.branch("branch-" + i + "-" + name, AgentAdapters.toFunction(ex));
+          }
+        });
+    return self();
+  }
+
   @Override
   public AgentTaskItemListBuilder self() {
     return this;

fluent/agentic/src/main/java/io/serverlessworkflow/fluent/agentic/DelegatingAgentDoTaskFluent.java

@@ -42,8 +42,16 @@ default SELF sequence(String name, Object... agents) {
   }
 
   @SuppressWarnings("unchecked")
+  @Override
   default SELF loop(String name, Consumer<LoopAgentsBuilder> consumer) {
     d().loop(name, consumer);
     return (SELF) this;
   }
+
+  @SuppressWarnings("unchecked")
+  @Override
+  default SELF parallel(String name, Object... agents) {
+    d().parallel(name, agents);
+    return (SELF) this;
+  }
 }

fluent/agentic/src/test/java/io/serverlessworkflow/fluent/agentic/AgentTaskItemListBuilderTest.java

@@ -61,12 +61,12 @@ void testSequence() {
     assertThat(items.get(1).getName()).isEqualTo("seq-1");
     assertThat(items.get(2).getName()).isEqualTo("seq-2");
 
-    // All must be call tasks
+    // All must be call branche
     items.forEach(it -> assertThat(it.getTask().getCallTask().get()).isNotNull());
   }
 
   @Test
-  @DisplayName("loop(name, builder) produces a ForTaskFunction with inner call tasks")
+  @DisplayName("loop(name, builder) produces a ForTaskFunction with inner call branche")
   void testLoop() {
     AgentTaskItemListBuilder b = new AgentTaskItemListBuilder();
     Agents.MovieExpert scorer = AgentsUtils.newMovieExpert();

fluent/agentic/src/test/java/io/serverlessworkflow/fluent/agentic/AgenticWorkflowBuilderTest.java

@@ -25,13 +25,15 @@
 
 import dev.langchain4j.agentic.AgentServices;
 import dev.langchain4j.agentic.Cognisphere;
+import io.serverlessworkflow.api.types.ForkTask;
 import io.serverlessworkflow.api.types.Task;
 import io.serverlessworkflow.api.types.TaskItem;
 import io.serverlessworkflow.api.types.Workflow;
 import io.serverlessworkflow.api.types.func.CallJava;
 import io.serverlessworkflow.api.types.func.ForTaskFunction;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Predicate;
+import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 
 public class AgenticWorkflowBuilderTest {
@@ -142,4 +144,32 @@ void loopWithMaxIterationsAndExitCondition() {
     assertNotNull(fn.getWhilePredicate(), "While predicate set from exitCondition");
     assertThat(fn.getDo()).hasSize(1);
   }
+
+  @Test
+  @DisplayName("parallel() creates one ForkTask with N callFn branches")
+  void parallelAgents() {
+    Agents.MovieExpert a1 = AgentsUtils.newMovieExpert();
+    Agents.MovieExpert a2 = AgentsUtils.newMovieExpert();
+    Agents.MovieExpert a3 = AgentsUtils.newMovieExpert();
+
+    Workflow wf =
+        AgenticWorkflowBuilder.workflow("parallelFlow")
+            .tasks(d -> d.parallel("p", a1, a2, a3))
+            .build();
+
+    assertThat(wf.getDo()).hasSize(1);
+    TaskItem top = wf.getDo().get(0);
+    Task task = top.getTask();
+    assertThat(task.getForkTask()).isInstanceOf(ForkTask.class);
+
+    ForkTask fork = task.getForkTask();
+    assertThat(fork.getFork().getBranches()).hasSize(3);
+
+    fork.getFork()
+        .getBranches()
+        .forEach(
+            branch -> {
+              assertThat(branch.getTask().getCallTask().get()).isInstanceOf(CallJava.class);
+            });
+  }
 }

fluent/func/src/main/java/io/serverlessworkflow/fluent/func/DelegatingFuncDoTaskFluent.java

@@ -71,4 +71,16 @@ default SELF switchFn(Consumer<FuncSwitchTaskBuilder> cfg) {
     d().switchFn(cfg);
     return self();
   }
+
+  @Override
+  default SELF forkFn(String name, Consumer<FuncForkTaskBuilder> cfg) {
+    d().forkFn(name, cfg);
+    return self();
+  }
+
+  @Override
+  default SELF forkFn(Consumer<FuncForkTaskBuilder> cfg) {
+    d().forkFn(cfg);
+    return self();
+  }
 }

fluent/func/src/main/java/io/serverlessworkflow/fluent/func/FuncDoTaskFluent.java

@@ -30,4 +30,8 @@ public interface FuncDoTaskFluent<SELF extends FuncDoTaskFluent<SELF>> {
   SELF switchFn(String name, Consumer<FuncSwitchTaskBuilder> cfg);
 
   SELF switchFn(Consumer<FuncSwitchTaskBuilder> cfg);
+
+  SELF forkFn(String name, Consumer<FuncForkTaskBuilder> cfg);
+
+  SELF forkFn(Consumer<FuncForkTaskBuilder> cfg);
 }

fluent/func/src/main/java/io/serverlessworkflow/fluent/func/FuncForkTaskBuilder.java

@@ -0,0 +1,74 @@
+/*
+ * 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.fluent.func;
+
+import io.serverlessworkflow.api.types.ForkTask;
+import io.serverlessworkflow.api.types.ForkTaskConfiguration;
+import io.serverlessworkflow.api.types.Task;
+import io.serverlessworkflow.api.types.TaskItem;
+import io.serverlessworkflow.api.types.func.CallJava;
+import io.serverlessworkflow.api.types.func.CallTaskJava;
+import io.serverlessworkflow.fluent.spec.TaskBaseBuilder;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+public class FuncForkTaskBuilder extends TaskBaseBuilder<FuncForkTaskBuilder>
+    implements FuncTransformations<FuncForkTaskBuilder> {
+
+  private final ForkTask forkTask;
+  private final List<TaskItem> items;
+
+  FuncForkTaskBuilder() {
+    this.forkTask = new ForkTask();
+    this.forkTask.setFork(new ForkTaskConfiguration());
+    this.items = new ArrayList<>();
+  }
+
+  @Override
+  protected FuncForkTaskBuilder self() {
+    return this;
+  }
+
+  public <T, V> FuncForkTaskBuilder branch(String name, Function<T, V> function) {
+    this.items.add(
+        new TaskItem(name, new Task().withCallTask(new CallTaskJava(CallJava.function(function)))));
+    return this;
+  }
+
+  public <T, V> FuncForkTaskBuilder branch(Function<T, V> function) {
+    return this.branch(UUID.randomUUID().toString(), function);
+  }
+
+  public FuncForkTaskBuilder branches(Consumer<FuncTaskItemListBuilder> consumer) {
+    final FuncTaskItemListBuilder builder = new FuncTaskItemListBuilder();
+    consumer.accept(builder);
+    this.items.addAll(builder.build());
+    return this;
+  }
+
+  public FuncForkTaskBuilder compete(boolean compete) {
+    this.forkTask.getFork().setCompete(compete);
+    return this;
+  }
+
+  public ForkTask build() {
+    this.forkTask.getFork().setBranches(this.items);
+    return forkTask;
+  }
+}

fluent/func/src/main/java/io/serverlessworkflow/fluent/func/FuncTaskItemListBuilder.java

@@ -82,4 +82,18 @@ public FuncTaskItemListBuilder switchFn(String name, Consumer<FuncSwitchTaskBuil
   public FuncTaskItemListBuilder switchFn(Consumer<FuncSwitchTaskBuilder> consumer) {
     return this.switchFn(UUID.randomUUID().toString(), consumer);
   }
+
+  @Override
+  public FuncTaskItemListBuilder forkFn(Consumer<FuncForkTaskBuilder> cfg) {
+    return this.forkFn(UUID.randomUUID().toString(), cfg);
+  }
+
+  @Override
+  public FuncTaskItemListBuilder forkFn(String name, Consumer<FuncForkTaskBuilder> cfg) {
+    this.requireNameAndConfig(name, cfg);
+    final FuncForkTaskBuilder forkTaskJavaBuilder = new FuncForkTaskBuilder();
+    cfg.accept(forkTaskJavaBuilder);
+    return this.addTaskItem(
+        new TaskItem(name, new Task().withForkTask(forkTaskJavaBuilder.build())));
+  }
 }

fluent/func/src/test/java/io/serverlessworkflow/fluent/func/JavaWorkflowBuilderTest.java

@@ -165,7 +165,7 @@ void testJavaFunctionalIO() {
     ForTaskFunction fn = (ForTaskFunction) forTaskFnHolder.getForTask();
     assertNotNull(fn);
 
-    // Inspect nested tasks inside the function loop
+    // Inspect nested branche inside the function loop
     List<TaskItem> nested = fn.getDo();
     assertEquals(1, nested.size());
     TaskBase nestedTask = nested.get(0).getTask().getSetTask();
@@ -219,7 +219,7 @@ void testCallJavaTask() {
   }
 
   @Test
-  @DisplayName("switchCaseFn (Java variant) coexists with spec tasks")
+  @DisplayName("switchCaseFn (Java variant) coexists with spec branche")
   void testSwitchCaseJava() {
     Workflow wf =
         FuncWorkflowBuilder.workflow("switchJava")