feat: add structured PARALLEL blocks with join-all tokens

Co-authored-by: Cursor <cursoragent@cursor.com>

ordo-core/src/main/java/com/jetlumen/ordo/core/DefaultOrdoEngine.java

@@ -35,8 +35,13 @@ import com.jetlumen.ordo.api.query.InstanceQuery;
 import com.jetlumen.ordo.api.query.Page;
 import com.jetlumen.ordo.api.query.PageRequest;
 import com.jetlumen.ordo.api.query.TaskQuery;
+import com.jetlumen.ordo.api.InstanceToken;
+import com.jetlumen.ordo.api.ParallelBranch;
+import com.jetlumen.ordo.api.ParallelMembership;
+import com.jetlumen.ordo.api.TokenStatus;
 import com.jetlumen.ordo.api.repository.ActionExecutionRepository;
 import com.jetlumen.ordo.api.repository.ApprovalTaskRepository;
+import com.jetlumen.ordo.api.repository.InstanceTokenRepository;
 import com.jetlumen.ordo.api.repository.ProcessDefinitionRepository;
 import com.jetlumen.ordo.api.repository.ProcessHistoryRepository;
 import com.jetlumen.ordo.api.repository.ProcessInstanceRepository;
@@ -75,6 +80,7 @@ public final class DefaultOrdoEngine implements OrdoEngine {
     private final ApprovalTaskRepository taskRepository;
     private final ProcessHistoryRepository historyRepository;
     private final ActionExecutionRepository actionExecutionRepository;
+    private final InstanceTokenRepository tokenRepository;
     private final List<OrdoEventListener> listeners;
     /** Monotonic millis offset so same-clock events keep causal order after JDBC Timestamp round-trips. */
     private final AtomicLong eventSequence = new AtomicLong();
@@ -86,6 +92,7 @@ public final class DefaultOrdoEngine implements OrdoEngine {
                              ApprovalTaskRepository taskRepository,
                              ProcessHistoryRepository historyRepository,
                              ActionExecutionRepository actionExecutionRepository,
+                             InstanceTokenRepository tokenRepository,
                              List<OrdoEventListener> listeners) {
         this.clock = Objects.requireNonNull(clock, "clock must not be null");
         this.assigneeResolver = Objects.requireNonNull(assigneeResolver, "assigneeResolver must not be null");
@@ -98,6 +105,7 @@ public final class DefaultOrdoEngine implements OrdoEngine {
         this.historyRepository = Objects.requireNonNull(historyRepository, "historyRepository must not be null");
         this.actionExecutionRepository = Objects.requireNonNull(actionExecutionRepository,
                 "actionExecutionRepository must not be null");
+        this.tokenRepository = Objects.requireNonNull(tokenRepository, "tokenRepository must not be null");
         this.listeners = List.copyOf(Objects.requireNonNull(listeners, "listeners must not be null"));
     }
 
@@ -222,16 +230,8 @@ public final class DefaultOrdoEngine implements OrdoEngine {
                 throw new InstanceAlreadyCompletedException(instanceId);
             }
             record(events, instanceId, null, null, eventType, actor, comment, now);
-            TaskAction action = new TaskAction(actor, comment, now);
-            for (ApprovalTask pending : taskRepository.findPendingByInstanceId(instanceId)) {
-                ApprovalTask skipped = new ApprovalTask(pending.id(), pending.instanceId(), pending.stepId(),
-                        pending.name(), pending.assignee(), TaskStatus.SKIPPED, pending.createdAt(), now, action,
-                        pending.dueAt());
-                if (taskRepository.completeIfPending(skipped)) {
-                    record(events, instanceId, pending.id(), pending.stepId(), ProcessEventType.TASK_SKIPPED, actor,
-                            comment, now);
-                }
-            }
+            skipRemainingPending(instanceId, actor, comment, now, events);
+            tokenRepository.completeActiveByInstanceId(instanceId);
             return finished;
         });
         dispatch(events);
@@ -409,6 +409,7 @@ public final class DefaultOrdoEngine implements OrdoEngine {
         List<ApprovalTask> siblings = taskRepository.findByInstanceIdAndStepId(instance.id(), overdue.stepId());
         skipPendingSiblings(siblings, "", null, now, events);
         enterStep(instance, definition, requireStep(definition, due.to()), now, queued, events);
+        touchToken(definition, instance.id(), due.to());
         record(events, overdue.instanceId(), overdue.id(), overdue.stepId(), ProcessEventType.TASK_ESCALATED, null,
                 due.to(), now);
         return true;
@@ -498,6 +499,9 @@ public final class DefaultOrdoEngine implements OrdoEngine {
                                    Instant now, List<PendingAction> queued, List<ProcessEvent> events) {
         StepTransition matched = resolveTransition(definition, step, instance);
         if (matched.toStepId() == null) {
+            if (completeBranchIfInner(instance, definition, step, now, queued, events)) {
+                return;
+            }
             completeInstance(instance, ProcessStatus.APPROVED, now, events);
         } else {
             enterStep(instance, definition, requireStep(definition, matched.toStepId()), now, queued, events);
@@ -508,8 +512,13 @@ public final class DefaultOrdoEngine implements OrdoEngine {
                            List<PendingAction> queued, List<ProcessEvent> events) {
         ApprovalStep current = start;
         for (int hops = 0; hops < MAX_CONSECUTIVE_ACTIONS; hops++) {
+            if (current.kind() == StepKind.PARALLEL) {
+                enterParallel(instance, definition, current, now, queued, events);
+                return;
+            }
             if (current.kind() == StepKind.APPROVAL) {
                 createStepTasks(instance, current, now, events);
+                touchToken(definition, instance.id(), current.id());
                 return;
             }
             String executionId = nextId();
@@ -518,8 +527,12 @@ public final class DefaultOrdoEngine implements OrdoEngine {
                     now.plusMillis(eventSequence.getAndIncrement()), null));
             queued.add(new PendingAction(executionId, current.actionKey(), instance.id(), current.id(),
                     instance.context()));
+            touchToken(definition, instance.id(), current.id());
             StepTransition matched = resolveTransition(definition, current, instance);
             if (matched.toStepId() == null) {
+                if (completeBranchIfInner(instance, definition, current, now, queued, events)) {
+                    return;
+                }
                 completeInstance(instance, ProcessStatus.APPROVED, now, events);
                 return;
             }
@@ -528,6 +541,44 @@ public final class DefaultOrdoEngine implements OrdoEngine {
         throw new IllegalStateException("too many consecutive action steps in instance: " + instance.id());
     }
 
+    private void enterParallel(ProcessInstance instance, ProcessDefinition definition, ApprovalStep parallel,
+                               Instant now, List<PendingAction> queued, List<ProcessEvent> events) {
+        record(events, instance.id(), null, parallel.id(), ProcessEventType.PARALLEL_ENTERED, null, parallel.id(), now);
+        for (var branch : parallel.branches()) {
+            String startId = branch.stepIds().get(0);
+            tokenRepository.insert(new InstanceToken(nextId(), instance.id(), parallel.id(), branch.id(), startId,
+                    TokenStatus.ACTIVE));
+            enterStep(instance, definition, requireStep(definition, startId), now, queued, events);
+        }
+    }
+
+    private boolean completeBranchIfInner(ProcessInstance instance, ProcessDefinition definition, ApprovalStep step,
+                                          Instant now, List<PendingAction> queued, List<ProcessEvent> events) {
+        ParallelMembership membership = definition.membershipOf(step.id()).orElse(null);
+        if (membership == null) {
+            return false;
+        }
+        ApprovalStep parallel = membership.parallel();
+        ParallelBranch branch = membership.branch();
+        tokenRepository.completeIfActive(instance.id(), parallel.id(), branch.id());
+        record(events, instance.id(), null, parallel.id(), ProcessEventType.BRANCH_COMPLETED, null, branch.id(), now);
+        List<InstanceToken> tokens = tokenRepository.findByInstanceAndParallel(instance.id(), parallel.id());
+        boolean joined = tokens.size() == parallel.branches().size()
+                && tokens.stream().allMatch(token -> token.status() == TokenStatus.COMPLETED);
+        if (joined) {
+            record(events, instance.id(), null, parallel.id(), ProcessEventType.PARALLEL_JOINED, null, parallel.id(),
+                    now);
+            advanceOrComplete(instance, definition, parallel, now, queued, events);
+        }
+        return true;
+    }
+
+    private void touchToken(ProcessDefinition definition, String instanceId, String stepId) {
+        definition.membershipOf(stepId).ifPresent(membership ->
+                tokenRepository.updateCurrentStepIfActive(instanceId, membership.parallel().id(),
+                        membership.branch().id(), stepId));
+    }
+
     private void finishCommittedWork(List<PendingAction> queued, List<ProcessEvent> events) {
         runQueuedActions(queued, events);
         dispatch(events);
@@ -625,6 +676,22 @@ public final class DefaultOrdoEngine implements OrdoEngine {
         ProcessEventType type = status == ProcessStatus.APPROVED
                 ? ProcessEventType.INSTANCE_APPROVED : ProcessEventType.INSTANCE_REJECTED;
         record(events, instance.id(), null, null, type, null, null, now);
+        skipRemainingPending(instance.id(), null, null, now, events);
+        tokenRepository.completeActiveByInstanceId(instance.id());
+    }
+
+    private void skipRemainingPending(String instanceId, String actor, String comment, Instant now,
+                                      List<ProcessEvent> events) {
+        TaskAction action = actor == null ? null : new TaskAction(actor, comment, now);
+        for (ApprovalTask pending : taskRepository.findPendingByInstanceId(instanceId)) {
+            ApprovalTask skipped = new ApprovalTask(pending.id(), pending.instanceId(), pending.stepId(),
+                    pending.name(), pending.assignee(), TaskStatus.SKIPPED, pending.createdAt(), now, action,
+                    pending.dueAt());
+            if (taskRepository.completeIfPending(skipped)) {
+                record(events, instanceId, pending.id(), pending.stepId(), ProcessEventType.TASK_SKIPPED, actor,
+                        comment, now);
+            }
+        }
     }
 
     private ProcessDefinition requireLatestDefinition(String definitionId) {

ordo-core/src/main/java/com/jetlumen/ordo/core/InMemoryOrdoEngine.java

@@ -17,6 +17,7 @@ import com.jetlumen.ordo.api.query.PageRequest;
 import com.jetlumen.ordo.api.query.TaskQuery;
 import com.jetlumen.ordo.core.repository.InMemoryActionExecutionRepository;
 import com.jetlumen.ordo.core.repository.InMemoryApprovalTaskRepository;
+import com.jetlumen.ordo.core.repository.InMemoryInstanceTokenRepository;
 import com.jetlumen.ordo.core.repository.InMemoryProcessDefinitionRepository;
 import com.jetlumen.ordo.core.repository.InMemoryProcessHistoryRepository;
 import com.jetlumen.ordo.core.repository.InMemoryProcessInstanceRepository;
@@ -72,6 +73,7 @@ public final class InMemoryOrdoEngine implements OrdoEngine {
                 new InMemoryApprovalTaskRepository(instanceRepository),
                 new InMemoryProcessHistoryRepository(),
                 new InMemoryActionExecutionRepository(),
+                new InMemoryInstanceTokenRepository(),
                 listeners);
     }
 

ordo-core/src/main/java/com/jetlumen/ordo/core/repository/InMemoryInstanceTokenRepository.java

@@ -0,0 +1,77 @@
+package com.jetlumen.ordo.core.repository;
+
+import com.jetlumen.ordo.api.InstanceToken;
+import com.jetlumen.ordo.api.TokenStatus;
+import com.jetlumen.ordo.api.repository.InstanceTokenRepository;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/** Development-only in-memory implementation of the parallel token port. */
+public final class InMemoryInstanceTokenRepository implements InstanceTokenRepository {
+    private final Map<String, InstanceToken> tokens = new LinkedHashMap<>();
+
+    @Override
+    public synchronized void insert(InstanceToken token) {
+        Objects.requireNonNull(token, "token must not be null");
+        if (tokens.putIfAbsent(key(token.instanceId(), token.parallelStepId(), token.branchId()), token) != null) {
+            throw new IllegalStateException("token already exists for branch: " + token.branchId());
+        }
+    }
+
+    @Override
+    public synchronized List<InstanceToken> findByInstanceId(String instanceId) {
+        return tokens.values().stream().filter(token -> token.instanceId().equals(instanceId)).toList();
+    }
+
+    @Override
+    public synchronized List<InstanceToken> findByInstanceAndParallel(String instanceId, String parallelStepId) {
+        return tokens.values().stream()
+                .filter(token -> token.instanceId().equals(instanceId) && token.parallelStepId().equals(parallelStepId))
+                .toList();
+    }
+
+    @Override
+    public synchronized boolean completeIfActive(String instanceId, String parallelStepId, String branchId) {
+        String key = key(instanceId, parallelStepId, branchId);
+        InstanceToken current = tokens.get(key);
+        if (current == null || current.status() != TokenStatus.ACTIVE) {
+            return false;
+        }
+        tokens.put(key, new InstanceToken(current.id(), current.instanceId(), current.parallelStepId(),
+                current.branchId(), current.currentStepId(), TokenStatus.COMPLETED));
+        return true;
+    }
+
+    @Override
+    public synchronized boolean updateCurrentStepIfActive(String instanceId, String parallelStepId, String branchId,
+                                                          String currentStepId) {
+        String key = key(instanceId, parallelStepId, branchId);
+        InstanceToken current = tokens.get(key);
+        if (current == null || current.status() != TokenStatus.ACTIVE) {
+            return false;
+        }
+        tokens.put(key, new InstanceToken(current.id(), current.instanceId(), current.parallelStepId(),
+                current.branchId(), currentStepId, TokenStatus.ACTIVE));
+        return true;
+    }
+
+    @Override
+    public synchronized void completeActiveByInstanceId(String instanceId) {
+        List<InstanceToken> snapshot = new ArrayList<>(tokens.values());
+        for (InstanceToken token : snapshot) {
+            if (token.instanceId().equals(instanceId) && token.status() == TokenStatus.ACTIVE) {
+                tokens.put(key(token.instanceId(), token.parallelStepId(), token.branchId()),
+                        new InstanceToken(token.id(), token.instanceId(), token.parallelStepId(), token.branchId(),
+                                token.currentStepId(), TokenStatus.COMPLETED));
+            }
+        }
+    }
+
+    private static String key(String instanceId, String parallelStepId, String branchId) {
+        return instanceId + '\0' + parallelStepId + '\0' + branchId;
+    }
+}

ordo-core/src/test/java/com/jetlumen/ordo/core/InMemoryOrdoEngineTest.java

@@ -6,6 +6,7 @@ import com.jetlumen.ordo.api.AssigneeResolver;
 import com.jetlumen.ordo.api.ApprovalPolicy;
 import com.jetlumen.ordo.api.ApprovalStep;
 import com.jetlumen.ordo.api.ApprovalTask;
+import com.jetlumen.ordo.api.ParallelBranch;
 import com.jetlumen.ordo.api.ProcessContext;
 import com.jetlumen.ordo.api.ProcessDefinition;
 import com.jetlumen.ordo.api.ProcessEvent;
@@ -853,6 +854,128 @@ class InMemoryOrdoEngineTest {
         assertThrows(IllegalArgumentException.class, () -> engine.processDue(0));
     }
 
+    @Test
+    void parallelJoinsAfterEveryBranchCompletes() {
+        engine.publish(parallelReview());
+        var instance = engine.start("parallel-review", "alice");
+        List<ApprovalTask> pending = engine.findPendingTasksByInstanceId(instance.id());
+        assertEquals(2, pending.size());
+        ApprovalTask legal = pending.stream().filter(task -> task.stepId().equals("legal")).findFirst().orElseThrow();
+        ApprovalTask finance = pending.stream().filter(task -> task.stepId().equals("finance")).findFirst().orElseThrow();
+
+        engine.approve(legal.id(), "lee");
+        assertEquals(ProcessStatus.RUNNING, engine.findInstance(instance.id()).orElseThrow().status());
+        assertEquals(1, engine.findPendingTasksByInstanceId(instance.id()).size());
+        assertEquals("finance", engine.findPendingTasksByInstanceId(instance.id()).get(0).stepId());
+
+        engine.approve(finance.id(), "fay");
+        ApprovalTask ceo = engine.findPendingTasksByInstanceId(instance.id()).get(0);
+        assertEquals("ceo", ceo.stepId());
+        engine.approve(ceo.id(), "cara");
+        assertEquals(ProcessStatus.APPROVED, engine.findInstance(instance.id()).orElseThrow().status());
+        List<ProcessEventType> types = engine.queryHistory(instance.id(), new PageRequest(0, 50)).content().stream()
+                .map(ProcessEvent::type).toList();
+        assertTrue(types.contains(ProcessEventType.PARALLEL_ENTERED));
+        assertTrue(types.contains(ProcessEventType.BRANCH_COMPLETED));
+        assertTrue(types.contains(ProcessEventType.PARALLEL_JOINED));
+    }
+
+    @Test
+    void parallelRejectSkipsTheOtherBranch() {
+        engine.publish(parallelReview());
+        var instance = engine.start("parallel-review", "alice");
+        ApprovalTask legal = engine.findPendingTasksByInstanceId(instance.id()).stream()
+                .filter(task -> task.stepId().equals("legal")).findFirst().orElseThrow();
+        ApprovalTask finance = engine.findPendingTasksByInstanceId(instance.id()).stream()
+                .filter(task -> task.stepId().equals("finance")).findFirst().orElseThrow();
+        engine.reject(legal.id(), "lee");
+        assertEquals(ProcessStatus.REJECTED, engine.findInstance(instance.id()).orElseThrow().status());
+        assertEquals(TaskStatus.SKIPPED, engine.findTask(finance.id()).orElseThrow().status());
+        assertTrue(engine.findPendingTasksByInstanceId(instance.id()).isEmpty());
+    }
+
+    @Test
+    void parallelWithdrawAndCancelSkipBranchTasks() {
+        engine.publish(parallelReview());
+        var withdrawn = engine.start("parallel-review", "alice");
+        engine.withdraw(withdrawn.id(), "alice");
+        assertEquals(ProcessStatus.WITHDRAWN, engine.findInstance(withdrawn.id()).orElseThrow().status());
+        assertTrue(engine.findPendingTasksByInstanceId(withdrawn.id()).isEmpty());
+
+        var cancelled = engine.start("parallel-review", "alice");
+        engine.cancel(cancelled.id(), "admin");
+        assertEquals(ProcessStatus.CANCELLED, engine.findInstance(cancelled.id()).orElseThrow().status());
+        assertTrue(engine.findPendingTasksByInstanceId(cancelled.id()).isEmpty());
+    }
+
+    @Test
+    void parallelBranchCanRunActionThenJoin() {
+        engine.publish(new ProcessDefinition("parallel-action", "P", List.of(
+                ApprovalStep.parallel("dept", "Dept", List.of(
+                        new ParallelBranch("legal", List.of("legal-mail")),
+                        new ParallelBranch("finance", List.of("finance")))),
+                ApprovalStep.action("legal-mail", "Legal mail", "ok-mail"),
+                ApprovalStep.single("finance", "Finance", "fay"),
+                ApprovalStep.single("ceo", "CEO", "cara")),
+                List.of(
+                        StepTransition.end("legal-mail"),
+                        StepTransition.end("finance"),
+                        StepTransition.always("dept", "ceo"),
+                        StepTransition.end("ceo"))));
+        var instance = engine.start("parallel-action", "alice");
+        assertEquals(1, engine.findPendingTasksByInstanceId(instance.id()).size());
+        engine.approve(engine.findPendingTasksByInstanceId(instance.id()).get(0).id(), "fay");
+        engine.approve(engine.findPendingTasksByInstanceId(instance.id()).get(0).id(), "cara");
+        assertEquals(ProcessStatus.APPROVED, engine.findInstance(instance.id()).orElseThrow().status());
+    }
+
+    @Test
+    void processDueGotoStaysInsideABranch() {
+        MutableClock clock = new MutableClock(Instant.parse("2026-01-15T09:00:00Z"));
+        InMemoryOrdoEngine dueEngine = new InMemoryOrdoEngine(clock);
+        dueEngine.publish(new ProcessDefinition("parallel-due", "P", List.of(
+                ApprovalStep.parallel("dept", "Dept", List.of(
+                        new ParallelBranch("legal", List.of("legal", "legal-2")),
+                        new ParallelBranch("finance", List.of("finance")))),
+                new ApprovalStep("legal", "Legal", List.of("lee"), ApprovalPolicy.ANY, StepKind.APPROVAL, null,
+                        StepDue.gotoStep(java.time.Duration.ofHours(1), "legal-2")),
+                ApprovalStep.single("legal-2", "Legal 2", "lisa"),
+                ApprovalStep.single("finance", "Finance", "fay"),
+                ApprovalStep.single("ceo", "CEO", "cara")),
+                List.of(
+                        StepTransition.end("legal"),
+                        StepTransition.end("legal-2"),
+                        StepTransition.end("finance"),
+                        StepTransition.always("dept", "ceo"),
+                        StepTransition.end("ceo"))));
+        var instance = dueEngine.start("parallel-due", "alice");
+        clock.set(Instant.parse("2026-01-15T10:00:00Z"));
+        assertEquals(1, dueEngine.processDue(10));
+        assertTrue(dueEngine.findPendingTasksByInstanceId(instance.id()).stream()
+                .anyMatch(task -> task.stepId().equals("legal-2")));
+        assertTrue(dueEngine.findPendingTasksByInstanceId(instance.id()).stream()
+                .anyMatch(task -> task.stepId().equals("finance")));
+        dueEngine.approve(dueEngine.findPendingTasksByAssignee("lisa").get(0).id(), "lisa");
+        dueEngine.approve(dueEngine.findPendingTasksByAssignee("fay").get(0).id(), "fay");
+        dueEngine.approve(dueEngine.findPendingTasksByAssignee("cara").get(0).id(), "cara");
+        assertEquals(ProcessStatus.APPROVED, dueEngine.findInstance(instance.id()).orElseThrow().status());
+    }
+
+    private static ProcessDefinition parallelReview() {
+        return new ProcessDefinition("parallel-review", "P", List.of(
+                ApprovalStep.parallel("dept", "Dept", List.of(
+                        new ParallelBranch("legal", List.of("legal")),
+                        new ParallelBranch("finance", List.of("finance")))),
+                ApprovalStep.single("legal", "Legal", "lee"),
+                ApprovalStep.single("finance", "Finance", "fay"),
+                ApprovalStep.single("ceo", "CEO", "cara")),
+                List.of(
+                        StepTransition.end("legal"),
+                        StepTransition.end("finance"),
+                        StepTransition.always("dept", "ceo"),
+                        StepTransition.end("ceo")));
+    }
+
     private static final class MutableClock extends Clock {
         private Instant instant;