feat(java): sub-workflow steps

sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java

@@ -5,6 +5,7 @@
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Base64;
 import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -408,7 +409,11 @@ private static Set<String> deferredNodes(List<Step> steps) {
             // "on_success" is the default/static path — only runtime-evaluated
             // conditions (on_failure / always / callable) force a deferred node.
             boolean runtimeCondition = step.condition != null && !"on_success".equals(step.condition);
-            if (step.fanOut != null || step.fanIn != null || step.gate != null || runtimeCondition) {
+            if (step.fanOut != null
+                    || step.fanIn != null
+                    || step.gate != null
+                    || step.subWorkflow != null
+                    || runtimeCondition) {
                 deferred.add(step.name);
             }
         }
@@ -436,7 +441,7 @@ public void cancelWorkflow(String runId) {
     }
 
     /** Encode a step's structure + per-step overrides for the native submit call. */
-    private static Map<String, Object> stepSpec(Step step) {
+    private Map<String, Object> stepSpec(Step step) {
         Map<String, Object> spec = new LinkedHashMap<>();
         spec.put("name", step.name);
         spec.put("taskName", step.taskName);
@@ -465,9 +470,50 @@ private static Map<String, Object> stepSpec(Step step) {
         if (step.condition != null) {
             spec.put("condition", step.condition);
         }
+        if (step.subWorkflow != null) {
+            spec.put("subWorkflow", encodeChild(step.subWorkflow));
+        }
         return spec;
     }
 
+    /**
+     * Encode a child workflow as a JSON string the worker tracker submits when the
+     * sub-workflow node is reached: its name/version, the child's step specs, its
+     * deferred set, and every child step's payload (Base64). Recurses through
+     * {@link #stepSpec}, so a child may itself contain gates/conditions/sub-workflows.
+     */
+    private String encodeChild(Workflow child) {
+        List<Step> steps = child.steps();
+        Set<String> deferred = deferredNodes(steps);
+        List<Map<String, Object>> specs = new ArrayList<>(steps.size());
+        Map<String, String> payloads = new LinkedHashMap<>();
+        for (Step step : steps) {
+            // A child run has no submit-time payload map and no callable registry, so
+            // any step that depends on either would lose state — reject it at build time.
+            if ("callable".equals(step.condition)) {
+                throw new TaskitoException("child workflow step '" + step.name
+                        + "' uses condition(Condition); a sub-workflow cannot carry callable predicates");
+            }
+            boolean controlNode =
+                    step.fanOut != null || step.fanIn != null || step.gate != null || step.subWorkflow != null;
+            if (!controlNode && step.payload == null) {
+                throw new TaskitoException("child workflow step '" + step.name
+                        + "' has no payload; a sub-workflow cannot supply child step payloads at submit");
+            }
+            specs.add(stepSpec(step));
+            if (step.payload != null) {
+                payloads.put(step.name, Base64.getEncoder().encodeToString(serializer.serialize(step.payload)));
+            }
+        }
+        Map<String, Object> blob = new LinkedHashMap<>();
+        blob.put("name", child.name());
+        blob.put("version", child.version());
+        blob.put("stepsJson", encode(specs));
+        blob.put("deferred", new ArrayList<>(deferred));
+        blob.put("payloads", payloads);
+        return encode(blob);
+    }
+
     /**
      * Encode a gate as a JSON string {@code {timeoutMs, onTimeout, message}}. The
      * core stores it opaquely on the step ({@code StepMetadata.gate}) and the

sdks/java/src/main/java/org/byteveda/taskito/workflows/PlanNode.java

@@ -20,6 +20,7 @@ final class PlanNode {
     final String fanIn;
     final String gate;
     final String condition;
+    final String subWorkflow;
 
     @JsonCreator
     PlanNode(
@@ -33,7 +34,8 @@ final class PlanNode {
             @JsonProperty("fanOut") String fanOut,
             @JsonProperty("fanIn") String fanIn,
             @JsonProperty("gate") String gate,
-            @JsonProperty("condition") String condition) {
+            @JsonProperty("condition") String condition,
+            @JsonProperty("subWorkflow") String subWorkflow) {
         this.name = name;
         this.predecessors = predecessors == null ? Collections.emptyList() : predecessors;
         this.taskName = taskName;
@@ -45,6 +47,7 @@ final class PlanNode {
         this.fanIn = fanIn;
         this.gate = gate;
         this.condition = condition;
+        this.subWorkflow = subWorkflow;
     }
 
     /** Whether this node is a fan-out/fan-in node (its job is created by the fan-out machinery). */

sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java

@@ -21,6 +21,7 @@ public final class Step {
     public final GateConfig gate;
     public final String condition;
     public final Condition callableCondition;
+    public final Workflow subWorkflow;
 
     private Step(Builder builder) {
         this.name = builder.name;
@@ -36,6 +37,7 @@ private Step(Builder builder) {
         this.gate = builder.gate;
         this.condition = builder.condition;
         this.callableCondition = builder.callableCondition;
+        this.subWorkflow = builder.subWorkflow;
     }
 
     /** Begin a step bound to a typed task. */
@@ -68,6 +70,7 @@ public static final class Builder {
         private GateConfig gate;
         private String condition;
         private Condition callableCondition;
+        private Workflow subWorkflow;
 
         private Builder(String name, String taskName, Object payload) {
             this.name = name;
@@ -177,6 +180,17 @@ public Builder condition(Condition predicate) {
             return this;
         }
 
+        /**
+         * Make this step a sub-workflow: instead of running a task it submits
+         * {@code child} as a child run and completes when the child finalizes
+         * (failing if the child fails). The running worker must
+         * {@code trackWorkflows(parent)}.
+         */
+        public Builder subWorkflow(Workflow child) {
+            this.subWorkflow = child;
+            return this;
+        }
+
         public Step build() {
             if (fanOut != null && fanIn != null) {
                 throw new IllegalArgumentException("step '" + name + "' cannot be both fan-out and fan-in");
@@ -191,6 +205,10 @@ public Step build() {
                         + "': a gate may only be created via Workflow.gate(...); a gate on a normal task step "
                         + "would defer the node and never enqueue its task");
             }
+            if (subWorkflow != null && (fanOut != null || fanIn != null || gate != null)) {
+                throw new IllegalArgumentException(
+                        "step '" + name + "' cannot be both a sub-workflow and a gate/fan-out/fan-in");
+            }
             return new Step(this);
         }
     }

sdks/java/src/main/java/org/byteveda/taskito/workflows/Workflow.java

@@ -93,6 +93,21 @@ public Workflow gate(String name, GateConfig gate, String... after) {
     /** Sentinel task name for gate control nodes (never enqueued). */
     static final String GATE_TASK = "__gate__";
 
+    /**
+     * Add a sub-workflow step: when reached it submits {@code child} as a child
+     * run and completes when the child finalizes (failing if the child fails).
+     * The running worker must {@code trackWorkflows(this)}.
+     */
+    public Workflow subWorkflow(String name, Workflow child, String... after) {
+        return step(Step.of(name, SUB_WORKFLOW_TASK, null)
+                .subWorkflow(child)
+                .after(after)
+                .build());
+    }
+
+    /** Sentinel task name for sub-workflow control nodes (never enqueued). */
+    static final String SUB_WORKFLOW_TASK = "__subworkflow__";
+
     // A fan-out/fan-in node has exactly one runtime trigger — its single producer.
     // Zero predecessors would never enqueue; multiple could fire from the wrong one.
     private static void requireSinglePredecessor(String kind, String name, String[] after) {

sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java

@@ -7,6 +7,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -51,6 +52,8 @@ public final class WorkflowTracker {
 
     /** Deferred-node payloads from registered workflows: {@code wfName -> node -> serialized payload}. */
     private final ConcurrentMap<String, Map<String, byte[]>> deferredPayloads = new ConcurrentHashMap<>();
+    /** Sub-workflow child runs' deferred-node payloads, keyed by run id so same-named runs don't collide. */
+    private final ConcurrentMap<String, Map<String, byte[]>> childRunPayloads = new ConcurrentHashMap<>();
     /** Callable conditions from registered workflows: {@code wfName -> node -> predicate}. */
     private final ConcurrentMap<String, Map<String, Condition>> callableConditions = new ConcurrentHashMap<>();
     /** Live gate timeout timers, so a manual resolution can cancel a pending auto-resolution. */
@@ -59,6 +62,8 @@ public final class WorkflowTracker {
     private final Set<GateKey> resolvedGates = ConcurrentHashMap.newKeySet();
     /** Deferred nodes already promoted (job created or gate parked), to dedupe concurrent outcomes. */
     private final Set<GateKey> promotedNodes = ConcurrentHashMap.newKeySet();
+    /** Sub-workflow child run → its parent {@code [runId, nodeName]}, so a finalized child resolves its parent node. */
+    private final ConcurrentMap<String, String[]> childToParent = new ConcurrentHashMap<>();
 
     public WorkflowTracker(QueueBackend backend, Serializer serializer) {
         this.backend = backend;
@@ -254,13 +259,50 @@ private void promoteOrSkip(String runId, PlanNode node, Map<String, NodeSnapshot
                 enterGate(runId, node);
                 return;
             }
+            if (node.subWorkflow != null) {
+                submitSubWorkflow(runId, node);
+                return;
+            }
             createDeferredJobFor(runId, node);
         } catch (RuntimeException e) {
             promotedNodes.remove(promotionKey);
             throw e;
         }
     }
 
+    /** Submit a sub-workflow node's child run and mark the node running until the child finalizes. */
+    private void submitSubWorkflow(String runId, PlanNode node) {
+        ChildSpec child = decode(node.subWorkflow, ChildSpec.class);
+        Set<String> deferred = new HashSet<>(child.deferred);
+        List<String> payloadNames = new ArrayList<>();
+        List<byte[]> payloadBytes = new ArrayList<>();
+        Map<String, byte[]> allPayloads = new HashMap<>();
+        child.payloads.forEach((nodeName, base64) -> {
+            byte[] bytes = java.util.Base64.getDecoder().decode(base64);
+            allPayloads.put(nodeName, bytes);
+            if (!deferred.contains(nodeName)) {
+                payloadNames.add(nodeName);
+                payloadBytes.add(bytes);
+            }
+        });
+        String childRun = backend.submitWorkflow(
+                child.name,
+                child.version,
+                child.stepsJson,
+                payloadNames.toArray(new String[0]),
+                payloadBytes.toArray(new byte[0][]),
+                null,
+                null,
+                deferred.toArray(new String[0]),
+                runId,
+                node.name);
+        // Register the child's payloads under its run id so the tracker can promote the
+        // child's own deferred nodes — run-scoped, so two same-named child runs can't stomp.
+        childRunPayloads.put(childRun, allPayloads);
+        childToParent.put(childRun, new String[] {runId, node.name});
+        backend.setWorkflowNodeRunning(runId, node.name);
+    }
+
     /** Whether {@code node}'s condition holds given its predecessors' outcomes. */
     private boolean shouldRun(String runId, PlanNode node, Map<String, NodeSnapshot> statuses) {
         if ("callable".equals(node.condition)) {
@@ -374,6 +416,10 @@ private void createDeferredJobFor(String runId, PlanNode node) {
     }
 
     private byte[] deferredPayload(String runId, String nodeName) {
+        Map<String, byte[]> childPayloads = childRunPayloads.get(runId);
+        if (childPayloads != null) {
+            return childPayloads.get(nodeName); // a child run resolves only its run-scoped payloads
+        }
         String wfName = workflowName(runId);
         if (wfName == null) {
             return null;
@@ -395,11 +441,32 @@ private String workflowName(String runId) {
     }
 
     private void finalizeIfTerminal(String runId) {
-        if (backend.finalizeRunIfTerminal(runId).isPresent()) {
-            forget(runId); // run reached a terminal state — drop its cached state
+        Optional<String> finalState = backend.finalizeRunIfTerminal(runId);
+        if (finalState.isEmpty()) {
+            return;
+        }
+        String[] parent = childToParent.remove(runId);
+        forget(runId); // run reached a terminal state — drop its cached state
+        if (parent != null) {
+            // This run is a sub-workflow child — resolve its parent node, then advance the parent.
+            boolean succeeded = "completed".equals(finalState.get());
+            resolveSubWorkflowParent(
+                    parent[0],
+                    parent[1],
+                    succeeded,
+                    succeeded ? null : "sub-workflow " + runId + " " + finalState.get());
         }
     }
 
+    private void resolveSubWorkflowParent(String parentRun, String parentNode, boolean succeeded, String error) {
+        backend.resolveWorkflowGate(parentRun, parentNode, succeeded, error);
+        RunPlan plan = plans.computeIfAbsent(parentRun, this::loadPlan);
+        if (plan != null) {
+            promoteSuccessors(parentRun, parentNode, plan);
+        }
+        finalizeIfTerminal(parentRun);
+    }
+
     /** Drop all per-run state once a run is terminal, cancelling any lingering timers. */
     private void forget(String runId) {
         plans.remove(runId);
@@ -416,6 +483,9 @@ private void forget(String runId) {
         });
         resolvedGates.removeIf(key -> key.runId.equals(runId));
         promotedNodes.removeIf(key -> key.runId.equals(runId));
+        childToParent.remove(runId);
+        childToParent.values().removeIf(parent -> parent[0].equals(runId));
+        childRunPayloads.remove(runId);
     }
 
     /** Stop the gate-timeout scheduler. Called when the owning worker closes. */
@@ -577,6 +647,30 @@ private static final class GateMeta {
         }
     }
 
+    /** A serialized child workflow carried in a sub-workflow node's metadata. */
+    @JsonIgnoreProperties(ignoreUnknown = true)
+    private static final class ChildSpec {
+        final String name;
+        final int version;
+        final String stepsJson;
+        final List<String> deferred;
+        final Map<String, String> payloads;
+
+        @JsonCreator
+        ChildSpec(
+                @JsonProperty("name") String name,
+                @JsonProperty("version") int version,
+                @JsonProperty("stepsJson") String stepsJson,
+                @JsonProperty("deferred") List<String> deferred,
+                @JsonProperty("payloads") Map<String, String> payloads) {
+            this.name = name;
+            this.version = version;
+            this.stepsJson = stepsJson;
+            this.deferred = deferred == null ? List.of() : deferred;
+            this.payloads = payloads == null ? Map.of() : payloads;
+        }
+    }
+
     /** {@code {runId, nodeName}} from the native node-ref view. */
     @JsonIgnoreProperties(ignoreUnknown = true)
     private static final class NodeRef {