From 80a92c1a707557e2dcb9b7022c865f4cad5c5b24 Mon Sep 17 00:00:00 2001
From: Mark Proctor <mdproctor@gmail.com>
Date: Wed, 8 Jul 2026 20:18:14 +0100
Subject: [PATCH 1/3] Replace serialization-based type inference with reified
 varargs type token
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The fluent DSL previously used SerializableFunction/Predicate/Consumer interfaces
and ReflectionUtils.inferInputType() to recover generic type information at runtime.
This relied on JVM serialization internals (writeReplace → SerializedLambda) which
are fragile, heavyweight, and force all lambdas to implement Serializable.

Replace with the reified varargs type token pattern: declaring a trailing T... parameter
causes Java to create a reified array at the call site, making the erased-but-concrete
class recoverable via getClass().getComponentType(). This gives the same runtime
Class<T> with zero reflection, zero serialization, zero allocation, and works with
any lambda or method reference regardless of serializability.

Changes:
- Replace ~40 call sites across FuncDSL, Step, CommonFuncOps, FuncEmitSpec,
  FuncEmitEventPropertiesBuilder, and FuncEventFilterSpec
- Delete ReflectionUtils, SerializableFunction, SerializablePredicate,
  SerializableConsumer (no longer needed)
- Drop Serializable from InstanceIdFunction and UniqueIdBiFunction (was only
  needed for the inference hack; ContextFunction and FilterFunction keep it
  as they are stored in serializable CallJava objects)
- Add test demonstrating nested generic type safety (Map<String, List<Integer>>)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Signed-off-by: Mark Proctor <mdproctor@gmail.com>
---
 .../func/FuncEmitEventPropertiesBuilder.java  |   3 +-
 .../fluent/func/dsl/CommonFuncOps.java        |  15 +-
 .../fluent/func/dsl/FuncDSL.java              | 285 +++++++++++-------
 .../fluent/func/dsl/FuncEmitSpec.java         |   8 +-
 .../fluent/func/dsl/FuncEventFilterSpec.java  |  10 +-
 .../fluent/func/dsl/Step.java                 | 103 +++----
 .../fluent/func/FuncDSLTest.java              |  20 ++
 .../reflection/func/InstanceIdFunction.java   |   4 +-
 .../api/reflection/func/ReflectionUtils.java  | 101 -------
 .../reflection/func/SerializableConsumer.java |  22 --
 .../reflection/func/SerializableFunction.java |  29 --
 .../func/SerializablePredicate.java           |  22 --
 .../reflection/func/UniqueIdBiFunction.java   |   3 +-
 13 files changed, 269 insertions(+), 356 deletions(-)
 delete mode 100644 experimental/types/src/main/java/io/serverlessworkflow/api/reflection/func/ReflectionUtils.java
 delete mode 100644 experimental/types/src/main/java/io/serverlessworkflow/api/reflection/func/SerializableConsumer.java
 delete mode 100644 experimental/types/src/main/java/io/serverlessworkflow/api/reflection/func/SerializableFunction.java
 delete mode 100644 experimental/types/src/main/java/io/serverlessworkflow/api/reflection/func/SerializablePredicate.java

diff --git a/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/FuncEmitEventPropertiesBuilder.java b/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/FuncEmitEventPropertiesBuilder.java
index 625b224af..6276f9e61 100644
--- a/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/FuncEmitEventPropertiesBuilder.java
+++ b/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;
   }
diff --git a/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/CommonFuncOps.java b/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/CommonFuncOps.java
index 202813c7b..a3df01d5b 100644
--- a/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/CommonFuncOps.java
+++ b/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")
+  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")
+  default <T> SwitchCaseSpec<T> caseOf(Predicate<T> when, T... typeToken) {
+    return caseOf(when, (Class<T>) typeToken.getClass().getComponentType());
   }
 
   default SwitchCaseConfigurer caseDefault(String task) {
diff --git a/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncDSL.java b/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncDSL.java
index 160d1b084..ca766e8ae 100644
--- a/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncDSL.java
+++ b/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncDSL.java
@@ -17,10 +17,6 @@
 
 import io.cloudevents.CloudEventData;
 import io.serverlessworkflow.api.reflection.func.InstanceIdFunction;
-import io.serverlessworkflow.api.reflection.func.ReflectionUtils;
-import io.serverlessworkflow.api.reflection.func.SerializableConsumer;
-import io.serverlessworkflow.api.reflection.func.SerializableFunction;
-import io.serverlessworkflow.api.reflection.func.SerializablePredicate;
 import io.serverlessworkflow.api.reflection.func.UniqueIdBiFunction;
 import io.serverlessworkflow.api.types.FlowDirectiveEnum;
 import io.serverlessworkflow.api.types.OAuth2AuthenticationData;
@@ -118,19 +114,18 @@ public static <T, V> Consumer<FuncCallTaskBuilder> fn(
    * @param <V> output type
    * @return a consumer that configures a {@code FuncCallTaskBuilder}
    */
-  public static <T, V> Consumer<FuncCallTaskBuilder> fn(SerializableFunction<T, V> function) {
-    return f ->
-        f.function(
-            function,
-            ReflectionUtils.inferInputType(function),
-            ReflectionUtils.inferResultType(function));
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public static <T, V> Consumer<FuncCallTaskBuilder> fn(Function<T, V> function, T... typeToken) {
+    Class<T> inputClass = (Class<T>) typeToken.getClass().getComponentType();
+    return OPS.fn(function, inputClass);
   }
 
   /**
    * Compose multiple switch cases into a single configurer for {@link FuncSwitchTaskBuilder}.
    *
-   * @param cases one or more {@link SwitchCaseConfigurer} built via {@link
-   *     #caseOf(SerializablePredicate)} or {@link #caseDefault(String)}
+   * @param cases one or more {@link SwitchCaseConfigurer} built via {@link #caseOf(Predicate,
+   *     Object[])} or {@link #caseDefault(String)}
    * @return a consumer to apply on a switch task builder
    */
   public static Consumer<FuncSwitchTaskBuilder> cases(SwitchCaseConfigurer... cases) {
@@ -156,8 +151,10 @@ public static <T> SwitchCaseSpec<T> caseOf(Predicate<T> when, Class<T> whenClass
    * @param <T> predicate input type
    * @return a fluent builder to set the consequent action (e.g., {@code then("taskName")})
    */
-  public static <T> SwitchCaseSpec<T> caseOf(SerializablePredicate<T> when) {
-    return OPS.caseOf(when);
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public static <T> SwitchCaseSpec<T> caseOf(Predicate<T> when, T... typeToken) {
+    return OPS.caseOf(when, (Class<T>) typeToken.getClass().getComponentType());
   }
 
   /**
@@ -241,15 +238,17 @@ public static FuncListenSpec toAny(String... types) {
    * @param <T> input type to the function
    * @return a consumer to configure {@link FuncEmitTaskBuilder}
    */
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
   public static <T> Consumer<FuncEmitTaskBuilder> produced(
-      String type, SerializableFunction<T, CloudEventData> function) {
-    return event ->
-        event.event(e -> e.type(type).data(function, ReflectionUtils.inferInputType(function)));
+      String type, Function<T, CloudEventData> function, T... typeToken) {
+    Class<T> inputClass = (Class<T>) typeToken.getClass().getComponentType();
+    return event -> event.event(e -> e.type(type).data(function, inputClass));
   }
 
   /**
-   * Same as {@link #produced(String, SerializableFunction)} but with an explicit input class to
-   * guide conversion.
+   * Same as {@link #produced(String, Function, Object[])} but with an explicit input class to guide
+   * conversion.
    *
    * @param type CloudEvent type
    * @param function function that maps workflow input to {@link CloudEventData}
@@ -262,16 +261,20 @@ public static <T> Consumer<FuncEmitTaskBuilder> produced(
     return event -> event.event(e -> e.type(type).data(function, inputClass));
   }
 
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
   public static <T> Consumer<FuncEmitTaskBuilder> produced(
-      String type, ContextFunction<T, CloudEventData> function) {
-    return event ->
-        event.event(e -> e.type(type).data(function, ReflectionUtils.inferInputType(function)));
+      String type, ContextFunction<T, CloudEventData> function, T... typeToken) {
+    Class<T> inputClass = (Class<T>) typeToken.getClass().getComponentType();
+    return event -> event.event(e -> e.type(type).data(function, inputClass));
   }
 
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
   public static <T> Consumer<FuncEmitTaskBuilder> produced(
-      String type, FilterFunction<T, CloudEventData> function) {
-    return event ->
-        event.event(e -> e.type(type).data(function, ReflectionUtils.inferInputType(function)));
+      String type, FilterFunction<T, CloudEventData> function, T... typeToken) {
+    Class<T> inputClass = (Class<T>) typeToken.getClass().getComponentType();
+    return event -> event.event(e -> e.type(type).data(function, inputClass));
   }
 
   /**
@@ -352,8 +355,10 @@ public static <T, R> FuncCallStep<T, R> withContext(ContextFunction<T, R> fn, Cl
     return withContext(null, fn, in);
   }
 
-  public static <T, R> FuncCallStep<T, R> withContext(ContextFunction<T, R> fn) {
-    return withContext(null, fn, ReflectionUtils.inferInputType(fn));
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public static <T, R> FuncCallStep<T, R> withContext(ContextFunction<T, R> fn, T... typeToken) {
+    return withContext(null, fn, (Class<T>) typeToken.getClass().getComponentType());
   }
 
   /**
@@ -368,7 +373,7 @@ public static <T, R> FuncCallStep<T, R> withContext(ContextFunction<T, R> fn) {
    */
   public static <T, R> FuncCallStep<T, R> withContext(
       String name, ContextFunction<T, R> fn, Class<T> in) {
-    return withContext(name, fn, in, ReflectionUtils.inferResultType(fn));
+    return withContext(name, fn, in, null);
   }
 
   public static <T, R> FuncCallStep<T, R> withContext(
@@ -376,8 +381,11 @@ public static <T, R> FuncCallStep<T, R> withContext(
     return new FuncCallStep<>(name, fn, in, out);
   }
 
-  public static <T, R> FuncCallStep<T, R> withContext(String name, ContextFunction<T, R> fn) {
-    return withContext(name, fn, ReflectionUtils.inferInputType(fn));
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public static <T, R> FuncCallStep<T, R> withContext(
+      String name, ContextFunction<T, R> fn, T... typeToken) {
+    return withContext(name, fn, (Class<T>) typeToken.getClass().getComponentType());
   }
 
   /**
@@ -409,7 +417,7 @@ public static <T, R> FuncCallStep<T, R> withFilter(FilterFunction<T, R> fn, Clas
    */
   public static <T, R> FuncCallStep<T, R> withFilter(
       String name, FilterFunction<T, R> fn, Class<T> in) {
-    return withFilter(name, fn, in, ReflectionUtils.inferResultType(fn));
+    return withFilter(name, fn, in, null);
   }
 
   public static <T, R> FuncCallStep<T, R> withFilter(
@@ -417,12 +425,17 @@ public static <T, R> FuncCallStep<T, R> withFilter(
     return new FuncCallStep<>(name, fn, in, out);
   }
 
-  public static <T, R> FuncCallStep<T, R> withFilter(FilterFunction<T, R> fn) {
-    return withFilter(null, fn, ReflectionUtils.inferInputType(fn));
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public static <T, R> FuncCallStep<T, R> withFilter(FilterFunction<T, R> fn, T... typeToken) {
+    return withFilter(null, fn, (Class<T>) typeToken.getClass().getComponentType());
   }
 
-  public static <T, R> FuncCallStep<T, R> withFilter(String name, FilterFunction<T, R> fn) {
-    return withFilter(name, fn, ReflectionUtils.inferInputType(fn));
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public static <T, R> FuncCallStep<T, R> withFilter(
+      String name, FilterFunction<T, R> fn, T... typeToken) {
+    return withFilter(name, fn, (Class<T>) typeToken.getClass().getComponentType());
   }
 
   /**
@@ -437,7 +450,7 @@ public static <T, R> FuncCallStep<T, R> withFilter(String name, FilterFunction<T
    */
   public static <T, R> FuncCallStep<T, R> withInstanceId(
       String name, InstanceIdFunction<T, R> fn, Class<T> in) {
-    return withInstanceId(name, fn, in, ReflectionUtils.inferResultType(fn));
+    return withInstanceId(name, fn, in, null);
   }
 
   public static <T, R> FuncCallStep<T, R> withInstanceId(
@@ -462,12 +475,18 @@ public static <T, R> FuncCallStep<T, R> withInstanceId(InstanceIdFunction<T, R>
     return withInstanceId(null, fn, in);
   }
 
-  public static <T, R> FuncCallStep<T, R> withInstanceId(String name, InstanceIdFunction<T, R> fn) {
-    return withInstanceId(name, fn, ReflectionUtils.inferInputType(fn));
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public static <T, R> FuncCallStep<T, R> withInstanceId(
+      String name, InstanceIdFunction<T, R> fn, T... typeToken) {
+    return withInstanceId(name, fn, (Class<T>) typeToken.getClass().getComponentType());
   }
 
-  public static <T, R> FuncCallStep<T, R> withInstanceId(InstanceIdFunction<T, R> fn) {
-    return withInstanceId(null, fn, ReflectionUtils.inferInputType(fn));
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public static <T, R> FuncCallStep<T, R> withInstanceId(
+      InstanceIdFunction<T, R> fn, T... typeToken) {
+    return withInstanceId(null, fn, (Class<T>) typeToken.getClass().getComponentType());
   }
 
   /**
@@ -516,13 +535,13 @@ static String defaultUniqueId(WorkflowContextData wctx, TaskContextData tctx) {
    * @param <R> result type
    * @return a named call step that can be chained with additional configurations
    * @see #withUniqueId(String, UniqueIdBiFunction, Class, Class)
-   * @see #withUniqueId(String, UniqueIdBiFunction)
+   * @see #withUniqueId(String, UniqueIdBiFunction, Object[])
    * @see #withUniqueId(UniqueIdBiFunction, Class)
-   * @see #withUniqueId(UniqueIdBiFunction)
+   * @see #withUniqueId(UniqueIdBiFunction, Object[])
    */
   public static <T, R> FuncCallStep<T, R> withUniqueId(
       String name, UniqueIdBiFunction<T, R> fn, Class<T> in) {
-    return withUniqueId(name, fn, in, ReflectionUtils.inferResultType(fn));
+    return withUniqueId(name, fn, in, null);
   }
 
   /**
@@ -561,9 +580,9 @@ public static <T, R> FuncCallStep<T, R> withUniqueId(
    * @param <R> result type
    * @return a named call step that can be chained with additional configurations
    * @see #withUniqueId(String, UniqueIdBiFunction, Class)
-   * @see #withUniqueId(String, UniqueIdBiFunction)
+   * @see #withUniqueId(String, UniqueIdBiFunction, Object[])
    * @see #withUniqueId(UniqueIdBiFunction, Class)
-   * @see #withUniqueId(UniqueIdBiFunction)
+   * @see #withUniqueId(UniqueIdBiFunction, Object[])
    */
   public static <T, R> FuncCallStep<T, R> withUniqueId(
       String name, UniqueIdBiFunction<T, R> fn, Class<T> in, Class<R> out) {
@@ -605,10 +624,13 @@ public static <T, R> FuncCallStep<T, R> withUniqueId(
    * @return a named call step that can be chained with additional configurations
    * @see #withUniqueId(String, UniqueIdBiFunction, Class)
    * @see #withUniqueId(String, UniqueIdBiFunction, Class, Class)
-   * @see #withUniqueId(UniqueIdBiFunction)
+   * @see #withUniqueId(UniqueIdBiFunction, Object[])
    */
-  public static <T, R> FuncCallStep<T, R> withUniqueId(String name, UniqueIdBiFunction<T, R> fn) {
-    return withUniqueId(name, fn, ReflectionUtils.inferInputType(fn));
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public static <T, R> FuncCallStep<T, R> withUniqueId(
+      String name, UniqueIdBiFunction<T, R> fn, T... typeToken) {
+    return withUniqueId(name, fn, (Class<T>) typeToken.getClass().getComponentType());
   }
 
   /**
@@ -644,7 +666,7 @@ public static <T, R> FuncCallStep<T, R> withUniqueId(String name, UniqueIdBiFunc
    * @return an unnamed call step that can be chained with additional configurations
    * @see #withUniqueId(String, UniqueIdBiFunction, Class)
    * @see #withUniqueId(String, UniqueIdBiFunction, Class, Class)
-   * @see #withUniqueId(UniqueIdBiFunction)
+   * @see #withUniqueId(UniqueIdBiFunction, Object[])
    */
   public static <T, R> FuncCallStep<T, R> withUniqueId(UniqueIdBiFunction<T, R> fn, Class<T> in) {
     return withUniqueId(null, fn, in);
@@ -678,12 +700,15 @@ public static <T, R> FuncCallStep<T, R> withUniqueId(UniqueIdBiFunction<T, R> fn
    * @param <T> input type (inferred automatically)
    * @param <R> result type (inferred automatically)
    * @return an unnamed call step that can be chained with additional configurations
-   * @see #withUniqueId(String, UniqueIdBiFunction)
+   * @see #withUniqueId(String, UniqueIdBiFunction, Object[])
    * @see #withUniqueId(UniqueIdBiFunction, Class)
    * @see #withUniqueId(String, UniqueIdBiFunction, Class)
    */
-  public static <T, R> FuncCallStep<T, R> withUniqueId(UniqueIdBiFunction<T, R> fn) {
-    return withUniqueId(null, fn, ReflectionUtils.inferInputType(fn));
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public static <T, R> FuncCallStep<T, R> withUniqueId(
+      UniqueIdBiFunction<T, R> fn, T... typeToken) {
+    return withUniqueId(null, fn, (Class<T>) typeToken.getClass().getComponentType());
   }
 
   /**
@@ -699,8 +724,10 @@ public static <T> ConsumeStep<T> consume(Consumer<T> consumer, Class<T> inputCla
     return new ConsumeStep<>(consumer, inputClass);
   }
 
-  public static <T> ConsumeStep<T> consume(SerializableConsumer<T> consumer) {
-    return consume(consumer, ReflectionUtils.inferInputType(consumer));
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public static <T> ConsumeStep<T> consume(Consumer<T> consumer, T... typeToken) {
+    return consume(consumer, (Class<T>) typeToken.getClass().getComponentType());
   }
 
   /**
@@ -716,8 +743,10 @@ public static <T> ConsumeStep<T> consume(String name, Consumer<T> consumer, Clas
     return new ConsumeStep<>(name, consumer, inputClass);
   }
 
-  public static <T> ConsumeStep<T> consume(String name, SerializableConsumer<T> consumer) {
-    return consume(name, consumer, ReflectionUtils.inferInputType(consumer));
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public static <T> ConsumeStep<T> consume(String name, Consumer<T> consumer, T... typeToken) {
+    return consume(name, consumer, (Class<T>) typeToken.getClass().getComponentType());
   }
 
   /**
@@ -737,8 +766,10 @@ public static <T, R> FuncCallStep<T, R> agent(UniqueIdBiFunction<T, R> fn, Class
     return withUniqueId(fn, in);
   }
 
-  public static <T, R> FuncCallStep<T, R> agent(UniqueIdBiFunction<T, R> fn) {
-    return withUniqueId(fn, ReflectionUtils.inferInputType(fn));
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public static <T, R> FuncCallStep<T, R> agent(UniqueIdBiFunction<T, R> fn, T... typeToken) {
+    return withUniqueId(fn, (Class<T>) typeToken.getClass().getComponentType());
   }
 
   /**
@@ -758,8 +789,11 @@ public static <T, R> FuncCallStep<T, R> agent(
     return withUniqueId(name, fn, in);
   }
 
-  public static <T, R> FuncCallStep<T, R> agent(String name, UniqueIdBiFunction<T, R> fn) {
-    return withUniqueId(name, fn, ReflectionUtils.inferInputType(fn));
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public static <T, R> FuncCallStep<T, R> agent(
+      String name, UniqueIdBiFunction<T, R> fn, T... typeToken) {
+    return withUniqueId(name, fn, (Class<T>) typeToken.getClass().getComponentType());
   }
 
   /**
@@ -801,13 +835,15 @@ public static <T, R> FuncCallStep<T, R> function(
    * @param <R> output type
    * @return a call step
    */
-  public static <T, R> FuncCallStep<T, R> function(SerializableFunction<T, R> fn) {
-    return new FuncCallStep<>(
-        fn, ReflectionUtils.inferInputType(fn), ReflectionUtils.inferResultType(fn));
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public static <T, R> FuncCallStep<T, R> function(Function<T, R> fn, T... typeToken) {
+    Class<T> inputClass = (Class<T>) typeToken.getClass().getComponentType();
+    return new FuncCallStep<>(fn, inputClass, null);
   }
 
   /**
-   * Named variant of {@link #function(SerializableFunction)} with inferred input type.
+   * Named variant of {@link #function(Function, Object[])} with inferred input type.
    *
    * @param name task name
    * @param fn the function to execute
@@ -815,9 +851,11 @@ public static <T, R> FuncCallStep<T, R> function(SerializableFunction<T, R> fn)
    * @param <R> output type
    * @return a named call step
    */
-  public static <T, R> FuncCallStep<T, R> function(String name, SerializableFunction<T, R> fn) {
-    return new FuncCallStep<>(
-        name, fn, ReflectionUtils.inferInputType(fn), ReflectionUtils.inferResultType(fn));
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
+  public static <T, R> FuncCallStep<T, R> function(String name, Function<T, R> fn, T... typeToken) {
+    Class<T> inputClass = (Class<T>) typeToken.getClass().getComponentType();
+    return new FuncCallStep<>(name, fn, inputClass, null);
   }
 
   /**
@@ -909,12 +947,12 @@ public static EmitStep emit(String name, Consumer<FuncEmitTaskBuilder> cfg) {
    * @param <T> input type
    * @return an {@link EmitStep}
    */
-  public static <T> EmitStep emit(String type, SerializableFunction<T, CloudEventData> fn) {
+  public static <T> EmitStep emit(String type, Function<T, CloudEventData> fn) {
     return new EmitStep(null, produced(type, fn));
   }
 
   /**
-   * Named variant of {@link #emit(String, SerializableFunction)}.
+   * Named variant of {@link #emit(String, Function)}.
    *
    * @param name task name
    * @param type CloudEvent type
@@ -922,8 +960,7 @@ public static <T> EmitStep emit(String type, SerializableFunction<T, CloudEventD
    * @param <T> input type
    * @return a named {@link EmitStep}
    */
-  public static <T> EmitStep emit(
-      String name, String type, SerializableFunction<T, CloudEventData> fn) {
+  public static <T> EmitStep emit(String name, String type, Function<T, CloudEventData> fn) {
     return new EmitStep(name, produced(type, fn));
   }
 
@@ -1004,7 +1041,7 @@ public static ListenStep listen(String name, FuncListenSpec spec) {
 
   /**
    * Low-level switch case configurer using a custom builder consumer. Prefer the {@link
-   * #caseOf(SerializablePredicate)} helpers when possible.
+   * #caseOf(Predicate, Object[])} helpers when possible.
    *
    * @param taskName optional task name
    * @param switchCase consumer to configure the {@link FuncSwitchTaskBuilder}
@@ -1026,8 +1063,8 @@ public static FuncTaskConfigurer switchCase(Consumer<FuncSwitchTaskBuilder> swit
   }
 
   /**
-   * Convenience to apply multiple {@link SwitchCaseConfigurer} built via {@link
-   * #caseOf(SerializablePredicate)}.
+   * Convenience to apply multiple {@link SwitchCaseConfigurer} built via {@link #caseOf(Predicate,
+   * Object[])}.
    *
    * @param cases case configurers
    * @return list configurer
@@ -1382,9 +1419,12 @@ public static <T> FuncTaskConfigurer switchWhenOrElse(
     return switchWhenOrElse(null, pred, thenTask, otherwise, predClass);
   }
 
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
   public static <T> FuncTaskConfigurer switchWhenOrElse(
-      SerializablePredicate<T> pred, String thenTask, FlowDirectiveEnum otherwise) {
-    return switchWhenOrElse(null, pred, thenTask, otherwise);
+      Predicate<T> pred, String thenTask, FlowDirectiveEnum otherwise, T... typeToken) {
+    return switchWhenOrElse(
+        null, pred, thenTask, otherwise, (Class<T>) typeToken.getClass().getComponentType());
   }
 
   /**
@@ -1410,13 +1450,16 @@ public static <T> FuncTaskConfigurer switchWhenOrElse(
             FuncDSL.cases(caseOf(pred, predClass).then(thenTask), caseDefault(otherwise)));
   }
 
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
   public static <T> FuncTaskConfigurer switchWhenOrElse(
       String taskName,
-      SerializablePredicate<T> pred,
+      Predicate<T> pred,
       String thenTask,
-      FlowDirectiveEnum otherwise) {
+      FlowDirectiveEnum otherwise,
+      T... typeToken) {
     return switchWhenOrElse(
-        taskName, pred, thenTask, otherwise, ReflectionUtils.inferInputType(pred));
+        taskName, pred, thenTask, otherwise, (Class<T>) typeToken.getClass().getComponentType());
   }
 
   /**
@@ -1434,9 +1477,12 @@ public static <T> FuncTaskConfigurer switchWhenOrElse(
     return switchWhenOrElse(null, pred, thenTask, otherwiseTask, predClass);
   }
 
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
   public static <T> FuncTaskConfigurer switchWhenOrElse(
-      SerializablePredicate<T> pred, String thenTask, String otherwiseTask) {
-    return switchWhenOrElse(null, pred, thenTask, otherwiseTask);
+      Predicate<T> pred, String thenTask, String otherwiseTask, T... typeToken) {
+    return switchWhenOrElse(
+        null, pred, thenTask, otherwiseTask, (Class<T>) typeToken.getClass().getComponentType());
   }
 
   /**
@@ -1461,10 +1507,16 @@ public static <T> FuncTaskConfigurer switchWhenOrElse(
             taskName, cases(caseOf(pred, predClass).then(thenTask), caseDefault(otherwiseTask)));
   }
 
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
   public static <T> FuncTaskConfigurer switchWhenOrElse(
-      String taskName, SerializablePredicate<T> pred, String thenTask, String otherwiseTask) {
+      String taskName, Predicate<T> pred, String thenTask, String otherwiseTask, T... typeToken) {
     return switchWhenOrElse(
-        taskName, pred, thenTask, otherwiseTask, ReflectionUtils.inferInputType(pred));
+        taskName,
+        pred,
+        thenTask,
+        otherwiseTask,
+        (Class<T>) typeToken.getClass().getComponentType());
   }
 
   /**
@@ -1558,13 +1610,18 @@ public static FuncTaskConfigurer switchWhenOrElse(
    * @param <T> input type for the collection function
    * @return list configurer
    */
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
   public static <T, V> FuncTaskConfigurer forEach(
-      SerializableFunction<T, Collection<V>> collection, Consumer<FuncTaskItemListBuilder> body) {
-    return forEach(null, collection, body);
+      Function<T, Collection<V>> collection,
+      Consumer<FuncTaskItemListBuilder> body,
+      T... typeToken) {
+    Class<T> inputClass = (Class<T>) typeToken.getClass().getComponentType();
+    return list -> list.forEach(null, j -> j.collection(collection, inputClass).tasks(body));
   }
 
   /**
-   * Named variant of {@link #forEach(SerializableFunction, Consumer)}.
+   * Named variant of {@link #forEach(Function, Consumer, Object[])}.
    *
    * @param taskName task name for the for-loop task
    * @param collection function that returns the collection to iterate
@@ -1572,41 +1629,59 @@ public static <T, V> FuncTaskConfigurer forEach(
    * @param <T> input type for the collection function
    * @return list configurer
    */
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
   public static <T, V> FuncTaskConfigurer forEach(
       String taskName,
-      SerializableFunction<T, Collection<V>> collection,
-      Consumer<FuncTaskItemListBuilder> body) {
-    return list ->
-        list.forEach(
-            taskName,
-            j -> j.collection(collection, ReflectionUtils.inferInputType(collection)).tasks(body));
+      Function<T, Collection<V>> collection,
+      Consumer<FuncTaskItemListBuilder> body,
+      T... typeToken) {
+    Class<T> inputClass = (Class<T>) typeToken.getClass().getComponentType();
+    return list -> list.forEach(taskName, j -> j.collection(collection, inputClass).tasks(body));
   }
 
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
   public static <T, V> FuncTaskConfigurer forEach(
-      SerializableFunction<T, Collection<V>> collection, LoopFunction<T, V, ?> function) {
-    return forEach(null, collection, function);
+      Function<T, Collection<V>> collection, LoopFunction<T, V, ?> function, T... typeToken) {
+    Class<T> inputClass = (Class<T>) typeToken.getClass().getComponentType();
+    return list -> list.forEach(null, j -> j.collection(collection, inputClass).tasks(function));
   }
 
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
   public static <T, V> FuncTaskConfigurer forEach(
       String taskName,
-      SerializableFunction<T, Collection<V>> collection,
-      LoopFunction<T, V, ?> function) {
+      Function<T, Collection<V>> collection,
+      LoopFunction<T, V, ?> function,
+      T... typeToken) {
+    Class<T> inputClass = (Class<T>) typeToken.getClass().getComponentType();
     return list ->
-        list.forEach(
-            taskName,
-            j ->
-                j.collection(collection, ReflectionUtils.inferInputType(collection))
-                    .tasks(function));
+        list.forEach(taskName, j -> j.collection(collection, inputClass).tasks(function));
   }
 
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
   public static <T, V> FuncTaskConfigurer forEachItem(
-      SerializableFunction<T, Collection<V>> collection, Function<V, ?> function) {
-    return forEachItem(null, collection, function);
+      Function<T, Collection<V>> collection, Function<V, ?> function, T... typeToken) {
+    Class<T> inputClass = (Class<T>) typeToken.getClass().getComponentType();
+    return list ->
+        list.forEach(
+            null, j -> j.collection(collection, inputClass).tasks((t, v) -> function.apply((V) v)));
   }
 
+  @SafeVarargs
+  @SuppressWarnings("unchecked")
   public static <T, V> FuncTaskConfigurer forEachItem(
-      String taskName, SerializableFunction<T, Collection<V>> collection, Function<V, ?> function) {
-    return forEach(taskName, collection, ((t, v) -> function.apply((V) v)));
+      String taskName,
+      Function<T, Collection<V>> collection,
+      Function<V, ?> function,
+      T... typeToken) {
+    Class<T> inputClass = (Class<T>) typeToken.getClass().getComponentType();
+    return list ->
+        list.forEach(
+            taskName,
+            j -> j.collection(collection, inputClass).tasks((t, v) -> function.apply((V) v)));
   }
 
   /**
diff --git a/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncEmitSpec.java b/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncEmitSpec.java
index 7b7f9beec..48ab8dd74 100644
--- a/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncEmitSpec.java
+++ b/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();
   }
diff --git a/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncEventFilterSpec.java b/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncEventFilterSpec.java
index 3b8c08236..963cc259e 100644
--- a/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncEventFilterSpec.java
+++ b/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(
diff --git a/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/Step.java b/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/Step.java
index 2f5055b09..0ce509c32 100644
--- a/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/Step.java
+++ b/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();
   }
 
diff --git a/experimental/fluent/func/src/test/java/io/serverlessworkflow/fluent/func/FuncDSLTest.java b/experimental/fluent/func/src/test/java/io/serverlessworkflow/fluent/func/FuncDSLTest.java
index 1f518e99c..def4f2595 100644
--- a/experimental/fluent/func/src/test/java/io/serverlessworkflow/fluent/func/FuncDSLTest.java
+++ b/experimental/fluent/func/src/test/java/io/serverlessworkflow/fluent/func/FuncDSLTest.java
@@ -668,4 +668,24 @@ 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());
+    assertNotNull(items.get(0).getTask().getCallTask(), "CallTask expected");
+  }
 }
diff --git a/experimental/types/src/main/java/io/serverlessworkflow/api/reflection/func/InstanceIdFunction.java b/experimental/types/src/main/java/io/serverlessworkflow/api/reflection/func/InstanceIdFunction.java
index 083456d4f..25efee030 100644
--- a/experimental/types/src/main/java/io/serverlessworkflow/api/reflection/func/InstanceIdFunction.java
+++ b/experimental/types/src/main/java/io/serverlessworkflow/api/reflection/func/InstanceIdFunction.java
@@ -15,8 +15,6 @@
  */
 package io.serverlessworkflow.api.reflection.func;
 
-import java.io.Serializable;
-
 /**
  * Functions that expect a workflow instance ID injection in runtime
  *
@@ -24,6 +22,6 @@
  * @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);
 }
diff --git a/experimental/types/src/main/java/io/serverlessworkflow/api/reflection/func/ReflectionUtils.java b/experimental/types/src/main/java/io/serverlessworkflow/api/reflection/func/ReflectionUtils.java
deleted file mode 100644
index 6b30fcdb2..000000000
--- a/experimental/types/src/main/java/io/serverlessworkflow/api/reflection/func/ReflectionUtils.java
+++ /dev/null
@@ -1,101 +0,0 @@
-/*
- * 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.api.reflection.func;
-
-import io.serverlessworkflow.api.types.func.ContextFunction;
-import io.serverlessworkflow.api.types.func.FilterFunction;
-import java.lang.invoke.MethodType;
-import java.lang.invoke.SerializedLambda;
-import java.lang.reflect.Method;
-import java.util.function.Function;
-
-/**
- * Specially used by {@link Function} parameters in the Java Function.
- *
- * @see <a href="https://www.baeldung.com/java-serialize-lambda">Serialize a Lambda in Java</a>
- */
-public final class ReflectionUtils {
-
-  private ReflectionUtils() {}
-
-  @SuppressWarnings("unchecked")
-  public static <T> Class<T> inferInputType(ContextFunction<T, ?> fn) {
-    return (Class<T>) inferInputTypeFromAny(fn, 0);
-  }
-
-  @SuppressWarnings("unchecked")
-  public static <T> Class<T> inferInputType(FilterFunction<T, ?> fn) {
-    return (Class<T>) inferInputTypeFromAny(fn, 0);
-  }
-
-  @SuppressWarnings("unchecked")
-  public static <T> Class<T> inferInputType(SerializableFunction<T, ?> fn) {
-    return (Class<T>) inferInputTypeFromAny(fn, 0);
-  }
-
-  @SuppressWarnings("unchecked")
-  public static <T> Class<T> inferInputType(SerializablePredicate<T> fn) {
-    return (Class<T>) inferInputTypeFromAny(fn, 0);
-  }
-
-  @SuppressWarnings("unchecked")
-  public static <T> Class<T> inferInputType(InstanceIdFunction<T, ?> fn) {
-    return (Class<T>) inferInputTypeFromAny(fn, 1);
-  }
-
-  @SuppressWarnings("unchecked")
-  public static <T> Class<T> inferInputType(UniqueIdBiFunction<T, ?> fn) {
-    return (Class<T>) inferInputTypeFromAny(fn, 1);
-  }
-
-  @SuppressWarnings("unchecked")
-  public static <T> Class<T> inferInputType(SerializableConsumer<T> fn) {
-    return (Class<T>) inferInputTypeFromAny(fn, 0);
-  }
-
-  @SuppressWarnings("unchecked")
-  public static <T> Class<T> inferResultType(Object fn) {
-    return (Class<T>) inferMethodType(fn).returnType();
-  }
-
-  /**
-   * Extracts the input type using the resolved interface signature. * @param fn The serializable
-   * lambda
-   *
-   * @param lambdaParamIndex The index of the payload parameter in the interface's apply method
-   */
-  public static Class<?> inferInputTypeFromAny(Object fn, int lambdaParamIndex) {
-    return inferMethodType(fn).parameterArray()[lambdaParamIndex];
-  }
-
-  private static MethodType inferMethodType(Object fn) {
-    try {
-      Method m = fn.getClass().getDeclaredMethod("writeReplace");
-      m.setAccessible(true);
-      SerializedLambda sl = (SerializedLambda) m.invoke(fn);
-
-      ClassLoader cl = fn.getClass().getClassLoader();
-
-      // getInstantiatedMethodType() provides the exact generic signature resolved
-      // by the compiler, completely bypassing captured variables and method kind switches!
-
-      return MethodType.fromMethodDescriptorString(sl.getInstantiatedMethodType(), cl);
-    } catch (ReflectiveOperationException ex) {
-      throw new IllegalStateException(
-          "Cannot infer type from lambda. Pass Class<T> or use a method reference.", ex);
-    }
-  }
-}
diff --git a/experimental/types/src/main/java/io/serverlessworkflow/api/reflection/func/SerializableConsumer.java b/experimental/types/src/main/java/io/serverlessworkflow/api/reflection/func/SerializableConsumer.java
deleted file mode 100644
index 60df61d0c..000000000
--- a/experimental/types/src/main/java/io/serverlessworkflow/api/reflection/func/SerializableConsumer.java
+++ /dev/null
@@ -1,22 +0,0 @@
-/*
- * 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.api.reflection.func;
-
-import java.io.Serializable;
-import java.util.function.Consumer;
-
-@FunctionalInterface
-public interface SerializableConsumer<T> extends Consumer<T>, Serializable {}
diff --git a/experimental/types/src/main/java/io/serverlessworkflow/api/reflection/func/SerializableFunction.java b/experimental/types/src/main/java/io/serverlessworkflow/api/reflection/func/SerializableFunction.java
deleted file mode 100644
index c4b2eee08..000000000
--- a/experimental/types/src/main/java/io/serverlessworkflow/api/reflection/func/SerializableFunction.java
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
- * 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.api.reflection.func;
-
-import java.io.Serializable;
-import java.util.function.Function;
-
-/**
- * Alternative to Function for our DSL to discover the input parameter class in runtime via
- * reflection.
- *
- * @param <T>
- * @param <R>
- */
-@FunctionalInterface
-public interface SerializableFunction<T, R> extends Function<T, R>, Serializable {}
diff --git a/experimental/types/src/main/java/io/serverlessworkflow/api/reflection/func/SerializablePredicate.java b/experimental/types/src/main/java/io/serverlessworkflow/api/reflection/func/SerializablePredicate.java
deleted file mode 100644
index a960b8db8..000000000
--- a/experimental/types/src/main/java/io/serverlessworkflow/api/reflection/func/SerializablePredicate.java
+++ /dev/null
@@ -1,22 +0,0 @@
-/*
- * 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.api.reflection.func;
-
-import java.io.Serializable;
-import java.util.function.Predicate;
-
-@FunctionalInterface
-public interface SerializablePredicate<T> extends Predicate<T>, Serializable {}
diff --git a/experimental/types/src/main/java/io/serverlessworkflow/api/reflection/func/UniqueIdBiFunction.java b/experimental/types/src/main/java/io/serverlessworkflow/api/reflection/func/UniqueIdBiFunction.java
index 4340cc3f8..4e1378c69 100644
--- a/experimental/types/src/main/java/io/serverlessworkflow/api/reflection/func/UniqueIdBiFunction.java
+++ b/experimental/types/src/main/java/io/serverlessworkflow/api/reflection/func/UniqueIdBiFunction.java
@@ -15,7 +15,6 @@
  */
 package io.serverlessworkflow.api.reflection.func;
 
-import java.io.Serializable;
 import java.util.function.BiFunction;
 
 /**
@@ -26,6 +25,6 @@
  * @param <R> The task result output
  */
 @FunctionalInterface
-public interface UniqueIdBiFunction<T, R> extends BiFunction<String, T, R>, Serializable {
+public interface UniqueIdBiFunction<T, R> extends BiFunction<String, T, R> {
   R apply(String uniqueId, T object);
 }

From 73f21058b8e5f32b2159a63a46a1b82e82c3f49f Mon Sep 17 00:00:00 2001
From: Mark Proctor <mdproctor@gmail.com>
Date: Thu, 9 Jul 2026 09:52:02 +0100
Subject: [PATCH 2/3] Fix async regression: default null returnClass to
 Object.class

AbstractJavaCallExecutor uses outputClass.isPresent() to decide whether
to run functions via CompletableFuture.supplyAsync (async) or
completedFuture (sync). Passing null for the return class caused
functions to run synchronously, breaking the async contract that
FuncCallAsyncTest.testReferencedFunctionCall relies on.

Default null returnClass to Object.class in FuncCallStep constructors
so the executor always takes the async path.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Signed-off-by: Mark Proctor <mdproctor@gmail.com>
---
 .../fluent/func/dsl/FuncCallStep.java                  | 10 +++++++---
 1 file changed, 7 insertions(+), 3 deletions(-)

diff --git a/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncCallStep.java b/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncCallStep.java
index 036534b8a..c2590975f 100644
--- a/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncCallStep.java
+++ b/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncCallStep.java
@@ -29,6 +29,10 @@ public final class FuncCallStep<T, R> extends Step<FuncCallStep<T, R>, FuncCallT
   private final ContextFunction<T, R> ctxFn;
   private final FilterFunction<T, R> filterFn;
   private final Class<T> argClass;
+  @SuppressWarnings("unchecked")
+  private static <R> Class<R> defaultReturnClass(Class<R> c) {
+    return c != null ? c : (Class<R>) Object.class;
+  }
   private final Class<R> returnClass;
 
   /** Function<T,R> variant (unnamed). */
@@ -43,7 +47,7 @@ public final class FuncCallStep<T, R> extends Step<FuncCallStep<T, R>, FuncCallT
     this.ctxFn = null;
     this.filterFn = null;
     this.argClass = argClass;
-    this.returnClass = returnClass;
+    this.returnClass = defaultReturnClass(returnClass);
   }
 
   /** ContextFunction<T,R> variant (unnamed). */
@@ -58,7 +62,7 @@ public final class FuncCallStep<T, R> extends Step<FuncCallStep<T, R>, FuncCallT
     this.ctxFn = ctxFn;
     this.filterFn = null;
     this.argClass = argClass;
-    this.returnClass = returnClass;
+    this.returnClass = defaultReturnClass(returnClass);
   }
 
   /** FilterFunction<T,R> variant (unnamed). */
@@ -74,7 +78,7 @@ public final class FuncCallStep<T, R> extends Step<FuncCallStep<T, R>, FuncCallT
     this.ctxFn = null;
     this.filterFn = filterFn;
     this.argClass = argClass;
-    this.returnClass = returnClass;
+    this.returnClass = defaultReturnClass(returnClass);
   }
 
   @Override

From 2e76c0c3ea2e7924f2e32ddf573421aa2514a545 Mon Sep 17 00:00:00 2001
From: Mark Proctor <mdproctor@gmail.com>
Date: Thu, 9 Jul 2026 10:24:19 +0100
Subject: [PATCH 3/3] Fix spotless formatting in FuncCallStep

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Signed-off-by: Mark Proctor <mdproctor@gmail.com>
---
 .../io/serverlessworkflow/fluent/func/dsl/FuncCallStep.java    | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncCallStep.java b/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncCallStep.java
index c2590975f..07226c01c 100644
--- a/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncCallStep.java
+++ b/experimental/fluent/func/src/main/java/io/serverlessworkflow/fluent/func/dsl/FuncCallStep.java
@@ -29,11 +29,12 @@ public final class FuncCallStep<T, R> extends Step<FuncCallStep<T, R>, FuncCallT
   private final ContextFunction<T, R> ctxFn;
   private final FilterFunction<T, R> filterFn;
   private final Class<T> argClass;
+  private final Class<R> returnClass;
+
   @SuppressWarnings("unchecked")
   private static <R> Class<R> defaultReturnClass(Class<R> c) {
     return c != null ? c : (Class<R>) Object.class;
   }
-  private final Class<R> returnClass;
 
   /** Function<T,R> variant (unnamed). */
   FuncCallStep(Function<T, R> fn, Class<T> argClass, Class<R> returnClass) {