Review comments

Signed-off-by: Matheus André <matheusandr2@gmail.com>

Author: fjtirado

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

@@ -355,10 +355,6 @@ public <T> T addMetadataIfAbsent(String key, Supplier<T> supplier) {
     return (T) additionalObjects.computeIfAbsent(key, k -> supplier.get());
   }
 
-  public Object computeCorrelationValue(String key, Object value) {
-    return additionalObjects.computeIfAbsent(key, k -> value);
-  }
-
   @Override
   public <T> Optional<T> findMetadata(String key, Class<T> objectClass) {
     Object value = additionalObjects.get(key);

impl/core/src/main/java/io/serverlessworkflow/impl/events/AbstractTypeConsumer.java

@@ -24,7 +24,6 @@
 import io.serverlessworkflow.impl.WorkflowApplication;
 import io.serverlessworkflow.impl.WorkflowContext;
 import io.serverlessworkflow.impl.WorkflowModel;
-import io.serverlessworkflow.impl.WorkflowModelFactory;
 import java.util.AbstractCollection;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -42,6 +41,8 @@ public abstract class AbstractTypeConsumer
 
   private static final Logger logger = LoggerFactory.getLogger(AbstractTypeConsumer.class);
 
+  private static final CloudEventPredicate ALWAYS_TRUE = (ce, wf, t) -> true;
+
   protected abstract void registerToAll(Consumer<CloudEvent> consumer);
 
   protected abstract void unregisterFromAll();
@@ -61,8 +62,7 @@ public TypeEventRegistrationBuilder listen(
         application.cloudEventPredicateFactory().build(application, properties);
     Collection<CloudEventPredicate> correlationPredicates =
         buildCorrelationPredicates(register.getCorrelate(), application);
-    return new TypeEventRegistrationBuilder(
-        type, cePredicate, correlationPredicates, application.modelFactory());
+    return new TypeEventRegistrationBuilder(type, cePredicate, correlationPredicates);
   }
 
   private Collection<CloudEventPredicate> buildCorrelationPredicates(
@@ -83,29 +83,24 @@ private Collection<CloudEventPredicate> buildCorrelationPredicates(
 
   @Override
   public Collection<TypeEventRegistrationBuilder> listenToAll(WorkflowApplication application) {
-    return List.of(
-        new TypeEventRegistrationBuilder(null, null, List.of(), application.modelFactory()));
+    return List.of(new TypeEventRegistrationBuilder(null, ALWAYS_TRUE, List.of()));
   }
 
   private static class CloudEventConsumer extends AbstractCollection<TypeEventRegistration>
       implements Consumer<CloudEvent> {
-    private final WorkflowModelFactory modelFactory;
     private Collection<TypeEventRegistration> registrations = new CopyOnWriteArrayList<>();
 
-    CloudEventConsumer(WorkflowModelFactory modelFactory) {
-      this.modelFactory = modelFactory;
-    }
-
     @Override
     public void accept(CloudEvent ce) {
       logger.debug("Received cloud event {}", ce);
       for (TypeEventRegistration registration : registrations) {
-        if (registration.predicate().test(ce, registration.workflow(), registration.task())) {
-          if (!testCorrelation(ce, registration)) {
-            continue;
-          }
-          registration.consumer().accept(ce);
+        if (!registration.predicate().test(ce, registration.workflow(), registration.task())) {
+          continue;
+        }
+        if (!testCorrelation(ce, registration)) {
+          continue;
         }
+        registration.consumer().accept(ce);
       }
     }
 
@@ -114,12 +109,14 @@ private boolean testCorrelation(CloudEvent ce, TypeEventRegistration registratio
       if (predicates.isEmpty()) {
         return true;
       }
-      WorkflowModel eventModel = null;
+      boolean hasModelAware =
+          predicates.stream().anyMatch(ModelAwareCloudEventPredicate.class::isInstance);
+      WorkflowModel eventModel =
+          hasModelAware
+              ? registration.workflow().definition().application().modelFactory().from(ce)
+              : null;
       for (CloudEventPredicate pred : predicates) {
         if (pred instanceof ModelAwareCloudEventPredicate ma) {
-          if (eventModel == null) {
-            eventModel = modelFactory.from(ce);
-          }
           if (!ma.test(eventModel, registration.workflow(), registration.task())) {
             return false;
           }
@@ -161,7 +158,7 @@ public TypeEventRegistration register(
       TaskContext task) {
     if (builder.type() == null) {
       registerToAll(ce);
-      return new TypeEventRegistration(null, ce, null, workflow, task);
+      return new TypeEventRegistration(null, ce, ALWAYS_TRUE, workflow, task);
     } else {
       TypeEventRegistration registration =
           new TypeEventRegistration(
@@ -175,7 +172,7 @@ public TypeEventRegistration register(
           .computeIfAbsent(
               registration.type(),
               k -> {
-                CloudEventConsumer consumer = new CloudEventConsumer(builder.modelFactory());
+                CloudEventConsumer consumer = new CloudEventConsumer();
                 register(k, consumer);
                 return consumer;
               })

impl/core/src/main/java/io/serverlessworkflow/impl/events/CorrelationPredicate.java

@@ -56,7 +56,12 @@ public static CorrelationPredicate from(
   }
 
   private String correlationStateKey(TaskContext task) {
-    return "correlation:" + task.position().jsonPointer() + ":" + correlationKey;
+    return "correlation:"
+        + task.position().jsonPointer()
+        + ":"
+        + task.iteration()
+        + ":"
+        + correlationKey;
   }
 
   @Override
@@ -75,7 +80,7 @@ public boolean test(WorkflowModel eventModel, WorkflowContext workflow, TaskCont
 
     if (expectResolver == null) {
       String stateKey = correlationStateKey(task);
-      Object firstValue = workflow.instance().computeCorrelationValue(stateKey, eventValue);
+      Object firstValue = workflow.instance().addMetadataIfAbsent(stateKey, () -> eventValue);
       boolean result = Objects.equals(eventValue, firstValue);
       logger.debug(
           "Correlation no expect, eventValue='{}', firstValue='{}', match={}",

impl/core/src/main/java/io/serverlessworkflow/impl/events/TypeEventRegistrationBuilder.java

@@ -15,18 +15,16 @@
  */
 package io.serverlessworkflow.impl.events;
 
-import io.serverlessworkflow.impl.WorkflowModelFactory;
 import java.util.Collection;
 import java.util.Collections;
 
 public record TypeEventRegistrationBuilder(
     String type,
     CloudEventPredicate cePredicate,
-    Collection<CloudEventPredicate> correlationPredicates,
-    WorkflowModelFactory modelFactory)
+    Collection<CloudEventPredicate> correlationPredicates)
     implements EventRegistrationBuilder {
 
   public TypeEventRegistrationBuilder(String type, CloudEventPredicate cePredicate) {
-    this(type, cePredicate, Collections.emptyList(), null);
+    this(type, cePredicate, Collections.emptyList());
   }
 }

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

@@ -83,8 +83,11 @@ void testCorrelateMatch(String sourceName, Workflow workflow) throws Exception {
                         Map.of("patientId", "P123", "name", "John"))));
 
     WorkflowModel result = future.get(2, TimeUnit.SECONDS);
-    List<Object> output = (List<Object>) JsonUtils.toJavaValue(JsonUtils.modelToJson(result));
+    Object outputValue = JsonUtils.toJavaValue(JsonUtils.modelToJson(result));
+    assertThat(outputValue).isInstanceOf(List.class);
+    List<?> output = (List<?>) outputValue;
     assertThat(output).hasSize(1);
+    assertThat(output.get(0)).isInstanceOf(Map.class);
     Map<String, Object> eventData = (Map<String, Object>) output.get(0);
     assertThat(eventData).containsEntry("patientId", "P123");
     assertThat(instance.status()).isEqualTo(WorkflowStatus.COMPLETED);
@@ -192,9 +195,16 @@ private static Workflow listenCorrelateNoExpectWorkflow() {
         .build();
   }
 
-  @Test
-  void testCorrelateNoExpectMatch() throws Exception {
-    Workflow workflow = listenCorrelateNoExpectWorkflow();
+  private static Stream<Arguments> correlateNoExpectWorkflowSources() throws IOException {
+    return Stream.of(
+            readWorkflowFromClasspath("workflows-samples/listen-correlate-no-expect.yaml"),
+            listenCorrelateNoExpectWorkflow())
+        .map(wf -> Arguments.of(wf.getDocument().getName(), wf));
+  }
+
+  @ParameterizedTest(name = "{0}")
+  @MethodSource("correlateNoExpectWorkflowSources")
+  void testCorrelateNoExpectMatch(String sourceName, Workflow workflow) throws Exception {
     WorkflowDefinition def = appl.workflowDefinition(workflow);
     WorkflowInstance instance = def.instance();
     CompletableFuture<WorkflowModel> future = instance.start();
@@ -213,16 +223,23 @@ void testCorrelateNoExpectMatch() throws Exception {
                         Map.of("patientId", "P123", "name", "John"))));
 
     WorkflowModel result = future.get(2, TimeUnit.SECONDS);
-    List<Object> output = (List<Object>) JsonUtils.toJavaValue(JsonUtils.modelToJson(result));
+    Object outputValue = JsonUtils.toJavaValue(JsonUtils.modelToJson(result));
+    assertThat(outputValue).isInstanceOf(List.class);
+    List<?> output = (List<?>) outputValue;
     assertThat(output).hasSize(1);
+    assertThat(output.get(0)).isInstanceOf(Map.class);
     Map<String, Object> eventData = (Map<String, Object>) output.get(0);
     assertThat(eventData).containsEntry("patientId", "P123");
     assertThat(instance.status()).isEqualTo(WorkflowStatus.COMPLETED);
   }
 
-  @Test
-  void testCorrelateNoExpectNoMatch() throws Exception {
-    Workflow workflow = listenCorrelateNoExpectWorkflow();
+  @ParameterizedTest(name = "{0}")
+  @MethodSource("correlateNoExpectWorkflowSources")
+  void testCorrelateNoExpectNoMatch(String sourceName, Workflow workflow) throws Exception {
+    assertCorrelateNoExpectNoMatch(workflow);
+  }
+
+  private void assertCorrelateNoExpectNoMatch(Workflow workflow) throws Exception {
     WorkflowDefinition def = appl.workflowDefinition(workflow);
     WorkflowInstance instance = def.instance();
     CompletableFuture<WorkflowModel> future = instance.start();

impl/test/src/test/resources/workflows-samples/listen-correlate-no-expect.yaml

@@ -0,0 +1,15 @@
+document:
+  dsl: '1.0.0-alpha5'
+  namespace: test
+  name: listen-correlate-no-expect
+  version: '0.1.0'
+do:
+  - waitForPatient:
+      listen:
+        to:
+          one:
+            with:
+              type: com.example.hospital.patient.admitted
+            correlate:
+              patientId:
+                from: .data.patientId

impl/test/src/test/resources/workflows-samples/listen-correlate.yaml

@@ -4,6 +4,8 @@ document:
   name: listen-correlate
   version: '0.1.0'
 input:
+  # Raw input shape: { "patientId": "P123" }
+  # After transform: { "patientId": "P123" } — same key, but explicitly shaped by input.from
   from: '{ patientId: .patientId }'
 do:
   - waitForPatient:
@@ -15,4 +17,6 @@ do:
             correlate:
               patientId:
                 from: .data.patientId
+                # expect is evaluated against the task input (post input.from transform),
+                # so .patientId resolves to the patientId field in the shaped input
                 expect: .patientId