feat(java): conditional workflow steps

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

@@ -398,14 +398,17 @@ public WorkflowRun submitWorkflow(Workflow workflow, Map<String, Object> supplie
     }
 
     /**
-     * Fan-out/fan-in and gate nodes — plus everything transitively downstream of
-     * them — are deferred: they carry a node row but no job at submit, and the
-     * worker tracker creates/parks them at runtime.
+     * Fan-out/fan-in, gate, and conditional nodes — plus everything transitively
+     * downstream of them — are deferred: they carry a node row but no job at
+     * submit, and the worker tracker creates/parks/evaluates them at runtime.
      */
     private static Set<String> deferredNodes(List<Step> steps) {
         Set<String> deferred = new HashSet<>();
         for (Step step : steps) {
-            if (step.fanOut != null || step.fanIn != null || step.gate != null) {
+            // "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) {
                 deferred.add(step.name);
             }
         }
@@ -459,6 +462,9 @@ private static Map<String, Object> stepSpec(Step step) {
         if (step.gate != null) {
             spec.put("gate", encodeGate(step.gate));
         }
+        if (step.condition != null) {
+            spec.put("condition", step.condition);
+        }
         return spec;
     }
 

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

@@ -0,0 +1,12 @@
+package org.byteveda.taskito.workflows;
+
+/**
+ * A predicate deciding whether a workflow step runs, given the run's state when
+ * its predecessors have settled. Registered with the running worker via
+ * {@code trackWorkflows(workflow)} — it is code, so it is not persisted; a
+ * workflow using a callable condition must be tracked on the worker that runs it.
+ */
+@FunctionalInterface
+public interface Condition {
+    boolean test(WorkflowContext context);
+}

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

@@ -19,6 +19,7 @@ final class PlanNode {
     final String fanOut;
     final String fanIn;
     final String gate;
+    final String condition;
 
     @JsonCreator
     PlanNode(
@@ -31,7 +32,8 @@ final class PlanNode {
             @JsonProperty("priority") Integer priority,
             @JsonProperty("fanOut") String fanOut,
             @JsonProperty("fanIn") String fanIn,
-            @JsonProperty("gate") String gate) {
+            @JsonProperty("gate") String gate,
+            @JsonProperty("condition") String condition) {
         this.name = name;
         this.predecessors = predecessors == null ? Collections.emptyList() : predecessors;
         this.taskName = taskName;
@@ -42,6 +44,7 @@ final class PlanNode {
         this.fanOut = fanOut;
         this.fanIn = fanIn;
         this.gate = gate;
+        this.condition = condition;
     }
 
     /** 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

@@ -19,6 +19,8 @@ public final class Step {
     public final String fanOut;
     public final String fanIn;
     public final GateConfig gate;
+    public final String condition;
+    public final Condition callableCondition;
 
     private Step(Builder builder) {
         this.name = builder.name;
@@ -32,6 +34,8 @@ private Step(Builder builder) {
         this.fanOut = builder.fanOut;
         this.fanIn = builder.fanIn;
         this.gate = builder.gate;
+        this.condition = builder.condition;
+        this.callableCondition = builder.callableCondition;
     }
 
     /** Begin a step bound to a typed task. */
@@ -62,6 +66,8 @@ public static final class Builder {
         private String fanOut;
         private String fanIn;
         private GateConfig gate;
+        private String condition;
+        private Condition callableCondition;
 
         private Builder(String name, String taskName, Object payload) {
             this.name = name;
@@ -127,6 +133,50 @@ public Builder gate(GateConfig gate) {
             return this;
         }
 
+        /**
+         * Run this step only when {@code condition} holds: {@code "on_success"}
+         * (every predecessor completed — the default), {@code "on_failure"} (any
+         * predecessor failed), or {@code "always"} (once predecessors settle). A
+         * conditional step is evaluated by the worker tracker, not pre-enqueued.
+         */
+        public Builder condition(String condition) {
+            if (condition != null
+                    && !"on_success".equals(condition)
+                    && !"on_failure".equals(condition)
+                    && !"always".equals(condition)) {
+                throw new IllegalArgumentException("unknown condition '" + condition
+                        + "'; use on_success, on_failure, always, or condition(Condition)");
+            }
+            this.condition = condition;
+            return this;
+        }
+
+        /** Run this step only if every predecessor completed (the default). */
+        public Builder onSuccess() {
+            return condition("on_success");
+        }
+
+        /** Run this step only if a predecessor failed (a recovery branch). */
+        public Builder onFailure() {
+            return condition("on_failure");
+        }
+
+        /** Run this step once predecessors settle, regardless of their outcome. */
+        public Builder always() {
+            return condition("always");
+        }
+
+        /**
+         * Run this step only when {@code predicate} holds. The predicate is code,
+         * so the workflow must be registered on the running worker via
+         * {@code trackWorkflows(workflow)}.
+         */
+        public Builder condition(Condition predicate) {
+            this.callableCondition = predicate;
+            this.condition = "callable";
+            return this;
+        }
+
         public Step build() {
             if (fanOut != null && fanIn != null) {
                 throw new IllegalArgumentException("step '" + name + "' cannot be both fan-out and fan-in");

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

@@ -0,0 +1,32 @@
+package org.byteveda.taskito.workflows;
+
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * The run's state passed to a callable {@link Condition} when deciding whether a
+ * step should run: the results and statuses of already-settled nodes.
+ *
+ * @param runId the workflow run id
+ * @param results deserialized results of completed nodes, by node name
+ * @param statuses every settled node's status, by node name
+ * @param successCount how many nodes completed
+ * @param failureCount how many nodes failed
+ */
+public record WorkflowContext(
+        String runId,
+        Map<String, Object> results,
+        Map<String, NodeStatus> statuses,
+        int successCount,
+        int failureCount) {
+
+    /** The result of {@code nodeName}, if it completed. */
+    public Optional<Object> result(String nodeName) {
+        return Optional.ofNullable(results.get(nodeName));
+    }
+
+    /** The status of {@code nodeName}, if it has settled. */
+    public Optional<NodeStatus> status(String nodeName) {
+        return Optional.ofNullable(statuses.get(nodeName));
+    }
+}