Add number type conversion validation to ensure proper casting between numeric types

- Implement convertNumber method in AbstractWorkflowModel to handle conversions between Integer, Long, Double, Float, Short, and Byte
- Add comprehensive test suite (WorkflowNumberConversionTest) covering all numeric type conversions
- Fix issue where asNumber() returned incompatible Number types without proper casting
- Ensure type safety when converting workflow model outputs to specific numeric types

Author: mcruzdev

impl/core/src/main/java/io/serverlessworkflow/impl/AbstractWorkflowModel.java

@@ -35,7 +35,7 @@ public <T> Optional<T> as(Class<T> clazz) {
     } else if (OffsetDateTime.class.isAssignableFrom(clazz)) {
       return (Optional<T>) asDate();
     } else if (Number.class.isAssignableFrom(clazz)) {
-      return (Optional<T>) asNumber();
+      return asNumber().map(num -> convertNumber(num, clazz));
     } else if (Collection.class.isAssignableFrom(clazz)) {
       Collection<?> collection = asCollection();
       return collection.isEmpty() ? Optional.empty() : (Optional<T>) Optional.of(collection);
@@ -45,4 +45,23 @@ public <T> Optional<T> as(Class<T> clazz) {
       return convert(clazz);
     }
   }
+
+  @SuppressWarnings("unchecked")
+  private <T> T convertNumber(Number num, Class<T> clazz) {
+    if (clazz == Integer.class) {
+      return (T) Integer.valueOf(num.intValue());
+    } else if (clazz == Long.class) {
+      return (T) Long.valueOf(num.longValue());
+    } else if (clazz == Double.class) {
+      return (T) Double.valueOf(num.doubleValue());
+    } else if (clazz == Float.class) {
+      return (T) Float.valueOf(num.floatValue());
+    } else if (clazz == Short.class) {
+      return (T) Short.valueOf(num.shortValue());
+    } else if (clazz == Byte.class) {
+      return (T) Byte.valueOf(num.byteValue());
+    } else {
+      return (T) num;
+    }
+  }
 }

impl/test/src/test/java/io/serverlessworkflow/impl/test/WorkflowNumberConversionTest.java

@@ -0,0 +1,163 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.serverlessworkflow.impl.test;
+
+import static io.serverlessworkflow.fluent.func.dsl.FuncDSL.function;
+
+import io.serverlessworkflow.api.types.Workflow;
+import io.serverlessworkflow.fluent.func.FuncWorkflowBuilder;
+import io.serverlessworkflow.impl.WorkflowApplication;
+import io.serverlessworkflow.impl.WorkflowModel;
+import java.util.function.Function;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class WorkflowNumberConversionTest {
+
+  @Test
+  void incompatible_test() {
+    Workflow workflow =
+        FuncWorkflowBuilder.workflow("numbers")
+            .tasks(
+                function(
+                        "scoreProposal",
+                        (Proposal input) -> {
+                          Integer score = calculateScore(input.abstractText());
+                          System.out.println("Score calculated having the result as: " + score);
+                          return score;
+                        },
+                        Proposal.class)
+                    .outputAs(
+                        (Integer score) -> new ProposalScore(score, score >= 7), Integer.class))
+            .build();
+
+    try (WorkflowApplication app = WorkflowApplication.builder().build()) {
+      WorkflowModel model =
+          app.workflowDefinition(workflow)
+              .instance(new Proposal("Workflow, workflow, workflow..."))
+              .start()
+              .join();
+      Assertions.assertNotNull(model);
+      ProposalScore result = model.as(ProposalScore.class).orElseThrow();
+      Assertions.assertEquals(10, result.score());
+      Assertions.assertTrue(result.accepted());
+    }
+  }
+
+  @Test
+  void long_to_integer_conversion() {
+    Workflow workflow =
+        FuncWorkflowBuilder.workflow("longToInt")
+            .tasks(
+                function("convertLong", Function.identity(), Long.class)
+                    .outputAs((Integer result) -> result * 2, Integer.class))
+            .build();
+
+    try (WorkflowApplication app = WorkflowApplication.builder().build()) {
+      WorkflowModel model = app.workflowDefinition(workflow).instance(100L).start().join();
+      Integer result = model.as(Integer.class).orElseThrow();
+      Assertions.assertEquals(200, result);
+    }
+  }
+
+  @Test
+  void integer_to_long_conversion() {
+    Workflow workflow =
+        FuncWorkflowBuilder.workflow("intToLong")
+            .tasks(
+                function("convertInt", Function.identity(), Integer.class)
+                    .outputAs((Long result) -> result * 3L, Long.class))
+            .build();
+
+    try (WorkflowApplication app = WorkflowApplication.builder().build()) {
+      WorkflowModel model = app.workflowDefinition(workflow).instance(50).start().join();
+      Long result = model.as(Long.class).orElseThrow();
+      Assertions.assertEquals(150L, result);
+    }
+  }
+
+  @Test
+  void double_to_integer_conversion() {
+    Workflow workflow =
+        FuncWorkflowBuilder.workflow("doubleToInt")
+            .tasks(
+                function("convertDouble", Function.identity(), Double.class)
+                    .outputAs((Integer result) -> result + 5, Integer.class))
+            .build();
+
+    try (WorkflowApplication app = WorkflowApplication.builder().build()) {
+      WorkflowModel model = app.workflowDefinition(workflow).instance(42.7).start().join();
+      Integer result = model.as(Integer.class).orElseThrow();
+      Assertions.assertEquals(47, result);
+    }
+  }
+
+  @Test
+  void float_to_double_conversion() {
+    Workflow workflow =
+        FuncWorkflowBuilder.workflow("floatToDouble")
+            .tasks(
+                function("convertFloat", Function.identity(), Float.class)
+                    .outputAs((Double result) -> result * 1.5, Double.class))
+            .build();
+
+    try (WorkflowApplication app = WorkflowApplication.builder().build()) {
+      WorkflowModel model = app.workflowDefinition(workflow).instance(10.0f).start().join();
+      Double result = model.as(Double.class).orElseThrow();
+      Assertions.assertEquals(15.0, result, 0.001);
+    }
+  }
+
+  @Test
+  void short_to_integer_conversion() {
+    Workflow workflow =
+        FuncWorkflowBuilder.workflow("shortToInt")
+            .tasks(
+                function("convertShort", (Short input) -> input.intValue(), Short.class)
+                    .outputAs((Integer result) -> result * 10, Integer.class))
+            .build();
+
+    try (WorkflowApplication app = WorkflowApplication.builder().build()) {
+      WorkflowModel model = app.workflowDefinition(workflow).instance((short) 5).start().join();
+      Integer result = model.as(Integer.class).orElseThrow();
+      Assertions.assertEquals(50, result);
+    }
+  }
+
+  @Test
+  void byte_to_integer_conversion() {
+    Workflow workflow =
+        FuncWorkflowBuilder.workflow("byteToInt")
+            .tasks(
+                function("convertByte", Function.identity(), Byte.class)
+                    .outputAs((Integer result) -> result + 100, Integer.class))
+            .build();
+
+    try (WorkflowApplication app = WorkflowApplication.builder().build()) {
+      WorkflowModel model = app.workflowDefinition(workflow).instance((byte) 25).start().join();
+      Integer result = model.as(Integer.class).orElseThrow();
+      Assertions.assertEquals(125, result);
+    }
+  }
+
+  private Integer calculateScore(String abstractText) {
+    return abstractText.contains("Workflow") ? 10 : 5;
+  }
+
+  public record ProposalScore(Integer score, boolean accepted) {}
+
+  public record Proposal(String abstractText) {}
+}