Replace serialization-based type inference with reified varargs type token

experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/FuncEmitEventPropertiesBuilder.java

@@ -16,7 +16,6 @@
 package io.serverlessworkflow.fluent.func;
 
 import io.cloudevents.CloudEventData;
-import io.serverlessworkflow.api.reflection.func.SerializableFunction;
 import io.serverlessworkflow.api.types.func.ContextFunction;
 import io.serverlessworkflow.api.types.func.EventDataFunction;
 import io.serverlessworkflow.api.types.func.FilterFunction;
@@ -31,7 +30,7 @@ protected FuncEmitEventPropertiesBuilder self() {
     return this;
   }
 
-  public <T> FuncEmitEventPropertiesBuilder data(SerializableFunction<T, CloudEventData> function) {
+  public <T> FuncEmitEventPropertiesBuilder data(Function<T, CloudEventData> function) {
     this.eventProperties.setData(new EventDataFunction().withFunction(function));
     return this;
   }

experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/CommonFuncOps.java

@@ -15,9 +15,6 @@
  */
 package io.serverlessworkflow.fluent.func.dsl;
 
-import io.serverlessworkflow.api.reflection.func.ReflectionUtils;
-import io.serverlessworkflow.api.reflection.func.SerializableFunction;
-import io.serverlessworkflow.api.reflection.func.SerializablePredicate;
 import io.serverlessworkflow.api.types.FlowDirectiveEnum;
 import io.serverlessworkflow.fluent.func.FuncCallTaskBuilder;
 import io.serverlessworkflow.fluent.func.FuncSwitchTaskBuilder;
@@ -32,9 +29,10 @@ default <T, V> Consumer<FuncCallTaskBuilder> fn(Function<T, V> function, Class<T
     return f -> f.function(function, argClass);
   }
 
-  default <T, V> Consumer<FuncCallTaskBuilder> fn(SerializableFunction<T, V> function) {
-    Class<T> clazz = ReflectionUtils.inferInputType(function);
-    return f -> f.function(function, clazz);
+  @SuppressWarnings({"unchecked", "varargs"})
+  default <T, V> Consumer<FuncCallTaskBuilder> fn(Function<T, V> function, T... typeToken) {
+    Class<T> clazz = (Class<T>) typeToken.getClass().getComponentType();
+    return fn(function, clazz);
   }
 
   default Consumer<FuncSwitchTaskBuilder> cases(SwitchCaseConfigurer... cases) {
@@ -49,8 +47,9 @@ default <T> SwitchCaseSpec<T> caseOf(Predicate<T> when, Class<T> whenClass) {
     return new SwitchCaseSpec<T>().when(when, whenClass);
   }
 
-  default <T> SwitchCaseSpec<T> caseOf(SerializablePredicate<T> when) {
-    return new SwitchCaseSpec<T>().when(when, ReflectionUtils.inferInputType(when));
+  @SuppressWarnings({"unchecked", "varargs"})
+  default <T> SwitchCaseSpec<T> caseOf(Predicate<T> when, T... typeToken) {
+    return caseOf(when, (Class<T>) typeToken.getClass().getComponentType());
   }
 
   default SwitchCaseConfigurer caseDefault(String task) {

experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncEmitSpec.java

@@ -18,8 +18,6 @@
 import io.cloudevents.CloudEventData;
 import io.cloudevents.core.data.BytesCloudEventData;
 import io.cloudevents.core.data.PojoCloudEventData;
-import io.serverlessworkflow.api.reflection.func.ReflectionUtils;
-import io.serverlessworkflow.api.reflection.func.SerializableFunction;
 import io.serverlessworkflow.api.types.func.ContextFunction;
 import io.serverlessworkflow.api.types.func.EventDataFunction;
 import io.serverlessworkflow.fluent.func.FuncEmitEventPropertiesBuilder;
@@ -40,8 +38,10 @@ protected FuncEmitSpec self() {
   }
 
   /** Sets the event data and the contentType to `application/json` */
-  public <T> FuncEmitSpec jsonData(SerializableFunction<T, CloudEventData> function) {
-    Class<T> clazz = ReflectionUtils.inferInputType(function);
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public final <T> FuncEmitSpec jsonData(Function<T, CloudEventData> function, T... typeToken) {
+    Class<T> clazz = (Class<T>) typeToken.getClass().getComponentType();
     addPropertyStep(e -> e.data(new EventDataFunction().withFunction(function, clazz)));
     return JSON();
   }

experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncEventFilterSpec.java

@@ -21,7 +21,6 @@
 import io.cloudevents.core.CloudEventUtils;
 import io.cloudevents.core.data.PojoCloudEventData;
 import io.cloudevents.jackson.PojoCloudEventDataMapper;
-import io.serverlessworkflow.api.reflection.func.SerializablePredicate;
 import io.serverlessworkflow.api.types.func.ContextPredicate;
 import io.serverlessworkflow.api.types.func.FilterPredicate;
 import io.serverlessworkflow.fluent.func.FuncEventFilterBuilder;
@@ -32,6 +31,7 @@
 import io.serverlessworkflow.impl.jackson.JsonUtils;
 import java.util.Map;
 import java.util.Objects;
+import java.util.function.Predicate;
 
 /**
  * Fluent DSL specification builder for configuring CloudEvent filters within a Serverless Workflow
@@ -58,7 +58,7 @@ protected FuncEventFilterSpec self() {
    * @param predicate the predicate to evaluate against the entire {@link CloudEvent}.
    * @return the current {@link FuncEventFilterSpec} instance.
    */
-  public FuncEventFilterSpec envelope(SerializablePredicate<CloudEvent> predicate) {
+  public FuncEventFilterSpec envelope(Predicate<CloudEvent> predicate) {
     addPropertyStep(e -> e.envelope(predicate));
     return this;
   }
@@ -95,7 +95,7 @@ public FuncEventFilterSpec envelope(FilterPredicate<CloudEvent> predicate) {
    * @param predicate the predicate to evaluate against the event data.
    * @return the current {@link FuncEventFilterSpec} instance.
    */
-  public FuncEventFilterSpec data(SerializablePredicate<CloudEventData> predicate) {
+  public FuncEventFilterSpec data(Predicate<CloudEventData> predicate) {
     addPropertyStep(e -> e.data(predicate));
     return this;
   }
@@ -139,7 +139,7 @@ public FuncEventFilterSpec data(FilterPredicate<CloudEventData> predicate) {
    * @param predicate the predicate to evaluate against the parsed Map.
    * @return the current {@link FuncEventFilterSpec} instance.
    */
-  public FuncEventFilterSpec dataAsMap(SerializablePredicate<Map<String, Object>> predicate) {
+  public FuncEventFilterSpec dataAsMap(Predicate<Map<String, Object>> predicate) {
     addPropertyStep(
         e ->
             e.envelope(
@@ -202,7 +202,7 @@ public FuncEventFilterSpec dataAsMap(FilterPredicate<Map<String, Object>> predic
    * @param <T> The target type.
    * @return the current {@link FuncEventFilterSpec} instance.
    */
-  public <T> FuncEventFilterSpec dataAs(Class<T> targetType, SerializablePredicate<T> predicate) {
+  public <T> FuncEventFilterSpec dataAs(Class<T> targetType, Predicate<T> predicate) {
     addPropertyStep(
         e ->
             e.envelope(

experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/Step.java

@@ -15,9 +15,6 @@
  */
 package io.serverlessworkflow.fluent.func.dsl;
 
-import io.serverlessworkflow.api.reflection.func.ReflectionUtils;
-import io.serverlessworkflow.api.reflection.func.SerializableFunction;
-import io.serverlessworkflow.api.reflection.func.SerializablePredicate;
 import io.serverlessworkflow.api.types.FlowDirectiveEnum;
 import io.serverlessworkflow.api.types.func.ContextFunction;
 import io.serverlessworkflow.api.types.func.FilterFunction;
@@ -51,11 +48,11 @@ protected SELF self() {
   // ---------------------------------------------------------------------------
 
   /** Queue a {@code when(predicate)} to be applied on the concrete builder. */
-  public <T> SELF when(SerializablePredicate<T> predicate) {
-    postConfigurers.add(
-        b ->
-            ((ConditionalTaskBuilder<?>) b)
-                .when(predicate, ReflectionUtils.inferInputType(predicate)));
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public final <T> SELF when(Predicate<T> predicate, T... typeToken) {
+    Class<T> inputClass = (Class<T>) typeToken.getClass().getComponentType();
+    postConfigurers.add(b -> ((ConditionalTaskBuilder<?>) b).when(predicate, inputClass));
     return self();
   }
 
@@ -127,11 +124,11 @@ public SELF then(FlowDirectiveEnum directive) {
    * @return this step for method chaining
    * @see io.serverlessworkflow.fluent.func.spi.FuncTaskTransformations#exportAs(Function)
    */
-  public <T, R> SELF exportAs(SerializableFunction<T, R> function) {
-    postConfigurers.add(
-        b ->
-            ((FuncTaskTransformations<?>) b)
-                .exportAs(function, ReflectionUtils.inferInputType(function)));
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public final <T, R> SELF exportAs(Function<T, R> function, T... typeToken) {
+    Class<T> inputClass = (Class<T>) typeToken.getClass().getComponentType();
+    postConfigurers.add(b -> ((FuncTaskTransformations<?>) b).exportAs(function, inputClass));
     return self();
   }
 
@@ -165,11 +162,11 @@ public <T, R> SELF exportAs(Function<T, R> function, Class<T> taskResultClass) {
    * @return this step for method chaining
    * @see io.serverlessworkflow.fluent.func.spi.FuncTaskTransformations#exportAs(FilterFunction)
    */
-  public <T, R> SELF exportAs(FilterFunction<T, R> function) {
-    postConfigurers.add(
-        b ->
-            ((FuncTaskTransformations<?>) b)
-                .exportAs(function, ReflectionUtils.inferInputType(function)));
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public final <T, R> SELF exportAs(FilterFunction<T, R> function, T... typeToken) {
+    Class<T> inputClass = (Class<T>) typeToken.getClass().getComponentType();
+    postConfigurers.add(b -> ((FuncTaskTransformations<?>) b).exportAs(function, inputClass));
     return self();
   }
 
@@ -202,11 +199,11 @@ public <T, R> SELF exportAs(FilterFunction<T, R> function, Class<T> taskResultCl
    * @return this step for method chaining
    * @see io.serverlessworkflow.fluent.func.spi.FuncTaskTransformations#exportAs(ContextFunction)
    */
-  public <T, R> SELF exportAs(ContextFunction<T, R> function) {
-    postConfigurers.add(
-        b ->
-            ((FuncTaskTransformations<?>) b)
-                .exportAs(function, ReflectionUtils.inferInputType(function)));
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public final <T, R> SELF exportAs(ContextFunction<T, R> function, T... typeToken) {
+    Class<T> inputClass = (Class<T>) typeToken.getClass().getComponentType();
+    postConfigurers.add(b -> ((FuncTaskTransformations<?>) b).exportAs(function, inputClass));
     return self();
   }
 
@@ -288,11 +285,11 @@ public SELF exportAs(String jqExpression) {
    * @return this step for method chaining
    * @see io.serverlessworkflow.fluent.func.spi.FuncTransformations#outputAs(Function)
    */
-  public <T, R> SELF outputAs(SerializableFunction<T, R> function) {
-    postConfigurers.add(
-        b ->
-            ((FuncTaskTransformations<?>) b)
-                .outputAs(function, ReflectionUtils.inferInputType(function)));
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public final <T, R> SELF outputAs(Function<T, R> function, T... typeToken) {
+    Class<T> inputClass = (Class<T>) typeToken.getClass().getComponentType();
+    postConfigurers.add(b -> ((FuncTaskTransformations<?>) b).outputAs(function, inputClass));
     return self();
   }
 
@@ -344,11 +341,11 @@ public <T, R> SELF outputAs(Function<T, R> function, Class<T> taskResultClass) {
    * @return this step for method chaining
    * @see io.serverlessworkflow.fluent.func.spi.FuncTransformations#outputAs(FilterFunction)
    */
-  public <T, R> SELF outputAs(FilterFunction<T, R> function) {
-    postConfigurers.add(
-        b ->
-            ((FuncTaskTransformations<?>) b)
-                .outputAs(function, ReflectionUtils.inferInputType(function)));
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public final <T, R> SELF outputAs(FilterFunction<T, R> function, T... typeToken) {
+    Class<T> inputClass = (Class<T>) typeToken.getClass().getComponentType();
+    postConfigurers.add(b -> ((FuncTaskTransformations<?>) b).outputAs(function, inputClass));
     return self();
   }
 
@@ -380,11 +377,11 @@ public <T, R> SELF outputAs(FilterFunction<T, R> function, Class<T> taskResultCl
    * @return this step for method chaining
    * @see io.serverlessworkflow.fluent.func.spi.FuncTransformations#outputAs(ContextFunction)
    */
-  public <T, R> SELF outputAs(ContextFunction<T, R> function) {
-    postConfigurers.add(
-        b ->
-            ((FuncTaskTransformations<?>) b)
-                .outputAs(function, ReflectionUtils.inferInputType(function)));
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public final <T, R> SELF outputAs(ContextFunction<T, R> function, T... typeToken) {
+    Class<T> inputClass = (Class<T>) typeToken.getClass().getComponentType();
+    postConfigurers.add(b -> ((FuncTaskTransformations<?>) b).outputAs(function, inputClass));
     return self();
   }
 
@@ -450,11 +447,11 @@ public SELF outputAs(String jqExpression) {
    * @return this step for method chaining
    * @see io.serverlessworkflow.fluent.func.spi.FuncTransformations#inputFrom(Function)
    */
-  public <T, R> SELF inputFrom(SerializableFunction<T, R> function) {
-    postConfigurers.add(
-        b ->
-            ((FuncTaskTransformations<?>) b)
-                .inputFrom(function, ReflectionUtils.inferInputType(function)));
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public final <T, R> SELF inputFrom(Function<T, R> function, T... typeToken) {
+    Class<T> inputClass = (Class<T>) typeToken.getClass().getComponentType();
+    postConfigurers.add(b -> ((FuncTaskTransformations<?>) b).inputFrom(function, inputClass));
     return self();
   }
 
@@ -509,11 +506,11 @@ public <T, R> SELF inputFrom(Function<T, R> function, Class<T> inputClass) {
    * @return this step for method chaining
    * @see io.serverlessworkflow.fluent.func.spi.FuncTransformations#inputFrom(FilterFunction)
    */
-  public <T, R> SELF inputFrom(FilterFunction<T, R> function) {
-    postConfigurers.add(
-        b ->
-            ((FuncTaskTransformations<?>) b)
-                .inputFrom(function, ReflectionUtils.inferInputType(function)));
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public final <T, R> SELF inputFrom(FilterFunction<T, R> function, T... typeToken) {
+    Class<T> inputClass = (Class<T>) typeToken.getClass().getComponentType();
+    postConfigurers.add(b -> ((FuncTaskTransformations<?>) b).inputFrom(function, inputClass));
     return self();
   }
 
@@ -547,11 +544,11 @@ public <T, R> SELF inputFrom(FilterFunction<T, R> function, Class<T> inputClass)
    * @return this step for method chaining
    * @see io.serverlessworkflow.fluent.func.spi.FuncTransformations#inputFrom(ContextFunction)
    */
-  public <T, R> SELF inputFrom(ContextFunction<T, R> function) {
-    postConfigurers.add(
-        b ->
-            ((FuncTaskTransformations<?>) b)
-                .inputFrom(function, ReflectionUtils.inferInputType(function)));
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public final <T, R> SELF inputFrom(ContextFunction<T, R> function, T... typeToken) {
+    Class<T> inputClass = (Class<T>) typeToken.getClass().getComponentType();
+    postConfigurers.add(b -> ((FuncTaskTransformations<?>) b).inputFrom(function, inputClass));
     return self();
   }
 

experimental/fluent/func/src/test/java/io/serverlessworkflow/fluent/func/FuncDSLTest.java

@@ -668,4 +668,31 @@ void grpc_unnamed_call_builds_task() {
     assertInstanceOf(CallGRPC.class, t.getCallTask().get());
     assertEquals("Call", ((CallGRPC) t.getCallTask().get()).getWith().getMethod());
   }
+
+  @Test
+  @DisplayName(
+      "function with nested generic Map<String, List<Integer>> preserves type safety in lambda")
+  void function_with_nested_generics_preserves_type_safety() {
+    Workflow wf =
+        FuncWorkflowBuilder.workflow("nested-generics")
+            .tasks(
+                function(
+                    (Map<String, List<Integer>> map) -> {
+                      List<Integer> values = map.get("scores");
+                      int total = values.stream().mapToInt(Integer::intValue).sum();
+                      return Map.of("total", total, "count", values.size());
+                    }))
+            .build();
+
+    List<TaskItem> items = wf.getDo();
+    assertEquals(1, items.size());
+
+    Task t = items.get(0).getTask();
+    assertNotNull(t.getCallTask(), "CallTask expected");
+    CallJava callJava = assertInstanceOf(CallJava.class, t.getCallTask().get());
+    assertEquals(
+        Map.class,
+        callJava.inputClass().orElse(null),
+        "Varargs type token should infer raw Map.class from Map<String, List<Integer>>");
+  }
 }

experimental/lambda/src/main/java/io/serverlessworkflow/impl/executors/func/AbstractJavaCallExecutor.java

@@ -65,11 +65,18 @@ public CompletableFuture<WorkflowModel> apply(
               workflowContext.definition().application().executorService())
           .thenApply(v -> output2Model(modelFactory, input, convertTypedResponse(v)));
     } else {
-      Object result =
-          convertResponse(callJavaFunction(workflowContext, taskContext, model2Input(input)));
-      return result instanceof CompletableFuture future
-          ? future.thenApply(v -> output2Model(modelFactory, input, convertResponse(v)))
-          : CompletableFuture.completedFuture(output2Model(modelFactory, input, result));
+      return CompletableFuture.supplyAsync(
+              () ->
+                  convertResponse(
+                      callJavaFunction(workflowContext, taskContext, model2Input(input))),
+              workflowContext.definition().application().executorService())
+          .thenCompose(
+              result ->
+                  result instanceof CompletableFuture<?> future
+                      ? future.thenApply(
+                          v -> output2Model(modelFactory, input, convertResponse(v)))
+                      : CompletableFuture.completedFuture(
+                          output2Model(modelFactory, input, result)));
     }
   }
 

experimental/types/src/main/java/io/serverlessworkflow/api/reflection/func/InstanceIdFunction.java

@@ -15,15 +15,13 @@
  */
 package io.serverlessworkflow.api.reflection.func;
 
-import java.io.Serializable;
-
 /**
  * Functions that expect a workflow instance ID injection in runtime
  *
  * @param <T> The task payload input
  * @param <R> The task result output
  */
 @FunctionalInterface
-public interface InstanceIdFunction<T, R> extends Serializable {
+public interface InstanceIdFunction<T, R> {
   R apply(String instanceId, T payload);
 }