From b8fbbe8de0b6f0cf6a0c629d859983125af84b3d Mon Sep 17 00:00:00 2001 From: Lucio Lelii Date: Wed, 22 Jul 2026 10:49:21 +0200 Subject: [PATCH] feat: add control-flow runtime foundations --- .../manager/executions/ExecutionContext.java | 18 +++- .../manager/executions/ExecutionListener.java | 6 ++ .../manager/executions/ExecutionObject.java | 3 +- .../executions/NodeExecutionResult.java | 30 ++++++ .../executions/executors/NodeExecutors.java | 75 +++++++++++-- .../persistence/ExecutionInputSnapshot.java | 2 + .../persistence/ExecutionSnapshot.java | 1 + .../persistence/ExecutionStepSnapshot.java | 2 + .../manager/executions/steps/Input.java | 49 +++++++-- .../steps/InputResolutionState.java | 8 ++ .../manager/executions/steps/Output.java | 6 ++ .../manager/executions/steps/Step.java | 67 ++++++++---- .../executions/steps/StepSkipReason.java | 7 ++ .../AllRequiredActivationPolicy.java | 32 ++++++ .../activation/StepActivationDecision.java | 10 ++ .../activation/StepActivationPolicy.java | 12 +++ .../StepActivationPolicyRegistry.java | 18 ++++ .../manager/executions/ExecutionTest.java | 74 +++++++++++++ .../executions/NodeExecutionResultTest.java | 27 +++++ ...nputResolutionAndActivationPolicyTest.java | 102 ++++++++++++++++++ 20 files changed, 504 insertions(+), 45 deletions(-) create mode 100644 src/main/java/it/cnr/isti/workflow/manager/executions/NodeExecutionResult.java create mode 100644 src/main/java/it/cnr/isti/workflow/manager/executions/steps/InputResolutionState.java create mode 100644 src/main/java/it/cnr/isti/workflow/manager/executions/steps/StepSkipReason.java create mode 100644 src/main/java/it/cnr/isti/workflow/manager/executions/steps/activation/AllRequiredActivationPolicy.java create mode 100644 src/main/java/it/cnr/isti/workflow/manager/executions/steps/activation/StepActivationDecision.java create mode 100644 src/main/java/it/cnr/isti/workflow/manager/executions/steps/activation/StepActivationPolicy.java create mode 100644 src/main/java/it/cnr/isti/workflow/manager/executions/steps/activation/StepActivationPolicyRegistry.java create mode 100644 src/test/java/it/cnr/isti/workflow/manager/executions/NodeExecutionResultTest.java create mode 100644 src/test/java/it/cnr/isti/workflow/manager/executions/steps/InputResolutionAndActivationPolicyTest.java diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionContext.java b/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionContext.java index 320e228..3188430 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionContext.java +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionContext.java @@ -15,6 +15,7 @@ import it.cnr.isti.workflow.manager.executions.persistence.ExecutionSnapshot; import it.cnr.isti.workflow.manager.executions.steps.Input; import it.cnr.isti.workflow.manager.executions.steps.Step; import it.cnr.isti.workflow.manager.executions.steps.StepStatus; +import it.cnr.isti.workflow.manager.executions.steps.StepSkipReason; import it.cnr.isti.workflow.manager.executions.bias.BiasExecutionContext; import it.cnr.isti.workflow.manager.llms.LLMDescriptor; import lombok.AccessLevel; @@ -217,15 +218,25 @@ public class ExecutionContext implements ExecutionListener { @Override public void skipped(String id) { + skipped(id, StepSkipReason.INPUT_UNAVAILABLE); + } + + @Override + public void skipped(String id, StepSkipReason reason) { if (this.status == ExecutionStatus.CANCELLED || this.status == ExecutionStatus.ERROR) { return; } logger.info("Step " + id + " skipped"); addStepEvent(id, ExecutionEventLevel.INFO, ExecutionEventType.STEP_SKIPPED, - "Skipped step " + resolveNodeName(id), Map.of()); + "Skipped step " + resolveNodeName(id), + reason == null ? Map.of() : Map.of("reason", reason.name())); Step skippedStep = this.steps.get(id); - skippedStep.getOutputs().forEach(output -> output.markUnavailable()); - skippedStep.getDependentSteps().forEach(dependentStep -> dependentStep.dependencySkipped(id)); + if (reason == StepSkipReason.BRANCH_NOT_SELECTED) { + skippedStep.getOutputs().forEach(output -> output.markNotSelected()); + } else { + skippedStep.getOutputs().forEach(output -> output.markUnavailable()); + } + skippedStep.getDependentSteps().forEach(dependentStep -> dependentStep.dependencySkipped(id, reason)); updateTerminalStatus(); notifyStateChanged(); } @@ -497,6 +508,7 @@ public class ExecutionContext implements ExecutionListener { public ExecutionSnapshot snapshot(Map providedAuthorizations) { return ExecutionSnapshot.builder() + .snapshotVersion(2) .status(this.status) .startTime(this.startTime) .endTime(this.endTime) diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionListener.java b/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionListener.java index 716b8cb..8a88f1a 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionListener.java +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionListener.java @@ -2,12 +2,18 @@ package it.cnr.isti.workflow.manager.executions; import java.util.Map; +import it.cnr.isti.workflow.manager.executions.steps.StepSkipReason; + public interface ExecutionListener { void completed(String id, Map result); void skipped(String id); + default void skipped(String id, StepSkipReason reason) { + skipped(id); + } + void failed(String id, String error); default void failed(String id, String error, String errorCode) { diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionObject.java b/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionObject.java index 799752e..ced2a08 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionObject.java +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionObject.java @@ -394,7 +394,8 @@ public class ExecutionObject { this.context.getSteps().values().forEach(step -> { switch (step.getStatus()) { case COMPLETED -> step.getDependentSteps().forEach(dependent -> dependent.dependencyCompleted(step.getId())); - case SKIPPED -> step.getDependentSteps().forEach(dependent -> dependent.dependencySkipped(step.getId())); + case SKIPPED -> step.getDependentSteps().forEach( + dependent -> dependent.dependencySkipped(step.getId(), step.getSkipReason())); default -> { } } diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/NodeExecutionResult.java b/src/main/java/it/cnr/isti/workflow/manager/executions/NodeExecutionResult.java new file mode 100644 index 0000000..ac2b2d2 --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/NodeExecutionResult.java @@ -0,0 +1,30 @@ +package it.cnr.isti.workflow.manager.executions; + +import java.util.LinkedHashSet; +import java.util.Map; +import java.util.Set; + +public record NodeExecutionResult( + Map outputs, + Set notSelectedOutputs) { + + public NodeExecutionResult { + outputs = outputs == null ? Map.of() : Map.copyOf(outputs); + notSelectedOutputs = notSelectedOutputs == null ? Set.of() : Set.copyOf(notSelectedOutputs); + Set overlap = new LinkedHashSet<>(outputs.keySet()); + overlap.retainAll(notSelectedOutputs); + if (!overlap.isEmpty()) { + throw new IllegalArgumentException("Outputs cannot be both produced and not selected: " + overlap); + } + } + + public static NodeExecutionResult completed(Map outputs) { + return new NodeExecutionResult(outputs, Set.of()); + } + + public static NodeExecutionResult routed(Map outputs, Set declaredOutputs) { + Set notSelected = new LinkedHashSet<>(declaredOutputs == null ? Set.of() : declaredOutputs); + notSelected.removeAll(outputs == null ? Set.of() : outputs.keySet()); + return new NodeExecutionResult(outputs, notSelected); + } +} diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/executors/NodeExecutors.java b/src/main/java/it/cnr/isti/workflow/manager/executions/executors/NodeExecutors.java index 892e388..2bb2255 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/executions/executors/NodeExecutors.java +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/executors/NodeExecutors.java @@ -2,10 +2,15 @@ package it.cnr.isti.workflow.manager.executions.executors; import java.util.List; import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; import it.cnr.isti.workflow.manager.blocks.Block; +import it.cnr.isti.workflow.manager.blocks.configurations.ConditionalBlockConfiguration; +import it.cnr.isti.workflow.manager.blocks.configurations.SwitchBlockConfiguration; import it.cnr.isti.workflow.manager.containers.Container; import it.cnr.isti.workflow.manager.executions.ExecutionEventLogger; +import it.cnr.isti.workflow.manager.executions.NodeExecutionResult; import it.cnr.isti.workflow.manager.executions.ExecutionVariableDescriptor; import it.cnr.isti.workflow.manager.executions.InteractionResult; import it.cnr.isti.workflow.manager.executions.bias.BiasExecutionContext; @@ -40,41 +45,77 @@ public final class NodeExecutors { public static Map execute(FlowNode node, List inputs, Map authorizations, Map executionVariables, Map executionVariableDescriptors, ExecutionEventLogger eventLogger) { - return execute(node, inputs, authorizations, executionVariables, executionVariableDescriptors, eventLogger, - BiasExecutionContext.normal()); + return executeResult(node, inputs, authorizations, executionVariables, executionVariableDescriptors, eventLogger) + .outputs(); } public static Map execute(FlowNode node, List inputs, Map authorizations, Map executionVariables, Map executionVariableDescriptors, ExecutionEventLogger eventLogger, BiasExecutionContext biasExecutionContext) { + return executeResult(node, inputs, authorizations, executionVariables, executionVariableDescriptors, eventLogger, + biasExecutionContext).outputs(); + } + + public static NodeExecutionResult executeResult(FlowNode node, List inputs, + Map authorizations, Map executionVariables, + Map executionVariableDescriptors, + ExecutionEventLogger eventLogger) { + return executeResult(node, inputs, authorizations, executionVariables, executionVariableDescriptors, eventLogger, + BiasExecutionContext.normal()); + } + + public static NodeExecutionResult executeResult(FlowNode node, List inputs, + Map authorizations, Map executionVariables, + Map executionVariableDescriptors, + ExecutionEventLogger eventLogger, BiasExecutionContext biasExecutionContext) { + Map outputs; if (node instanceof Block block) { - return executeBlock(block, inputs, authorizations, executionVariables, executionVariableDescriptors, + outputs = executeBlock(block, inputs, authorizations, executionVariables, executionVariableDescriptors, eventLogger, biasExecutionContext); - } - if (node instanceof Container container) { - return executeContainer(container, inputs, authorizations, executionVariables, executionVariableDescriptors, + } else if (node instanceof Container container) { + outputs = executeContainer(container, inputs, authorizations, executionVariables, executionVariableDescriptors, eventLogger, biasExecutionContext); + } else { + throw new IllegalStateException("No executor found for node type " + node.getClass().getName()); } - throw new IllegalStateException("No executor found for node type " + node.getClass().getName()); + return nodeExecutionResult(node, outputs); } public static Map simulate(FlowNode node, List inputs, Map authorizations, Map executionVariables, Map executionVariableDescriptors, LLMDescriptor simulatorDescriptor, ExecutionEventLogger eventLogger) { - return simulate(node, inputs, authorizations, executionVariables, executionVariableDescriptors, - simulatorDescriptor, eventLogger, BiasExecutionContext.normal()); + return simulateResult(node, inputs, authorizations, executionVariables, executionVariableDescriptors, + simulatorDescriptor, eventLogger).outputs(); } public static Map simulate(FlowNode node, List inputs, Map authorizations, Map executionVariables, Map executionVariableDescriptors, LLMDescriptor simulatorDescriptor, ExecutionEventLogger eventLogger, BiasExecutionContext biasExecutionContext) { + return simulateResult(node, inputs, authorizations, executionVariables, executionVariableDescriptors, + simulatorDescriptor, eventLogger, biasExecutionContext).outputs(); + } + + public static NodeExecutionResult simulateResult(FlowNode node, List inputs, + Map authorizations, Map executionVariables, + Map executionVariableDescriptors, + LLMDescriptor simulatorDescriptor, ExecutionEventLogger eventLogger, + BiasExecutionContext biasExecutionContext) { if (node instanceof Block block) { - return simulateBlock(block, inputs, authorizations, executionVariables, executionVariableDescriptors, - simulatorDescriptor, eventLogger, biasExecutionContext); + Map outputs = simulateBlock(block, inputs, authorizations, executionVariables, + executionVariableDescriptors, simulatorDescriptor, eventLogger, biasExecutionContext); + return nodeExecutionResult(node, outputs); } throw new IllegalStateException("No simulation executor found for node type " + node.getClass().getName()); } + public static NodeExecutionResult simulateResult(FlowNode node, List inputs, + Map authorizations, Map executionVariables, + Map executionVariableDescriptors, + LLMDescriptor simulatorDescriptor, ExecutionEventLogger eventLogger) { + return simulateResult(node, inputs, authorizations, executionVariables, executionVariableDescriptors, + simulatorDescriptor, eventLogger, BiasExecutionContext.normal()); + } + public static InteractionResult interact(FlowNode node, List inputs, Map interaction, Map partialResults, Map authorizations, Map executionVariables, Map executionVariableDescriptors, @@ -168,4 +209,16 @@ public final class NodeExecutors { BlockExecutors.get(block.getType()).cancel((Block) block, inputs, partialResults, authorizations, executionVariables, executionVariableDescriptors, eventLogger); } + + private static NodeExecutionResult nodeExecutionResult(FlowNode node, Map outputs) { + if (node instanceof Block block + && (block.getSpecificConfiguration() instanceof ConditionalBlockConfiguration + || block.getSpecificConfiguration() instanceof SwitchBlockConfiguration)) { + Set declaredOutputs = node.getOutputs().stream() + .map(output -> output.getName()) + .collect(Collectors.toCollection(java.util.LinkedHashSet::new)); + return NodeExecutionResult.routed(outputs, declaredOutputs); + } + return NodeExecutionResult.completed(outputs); + } } diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionInputSnapshot.java b/src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionInputSnapshot.java index 775dbd6..b542a74 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionInputSnapshot.java +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionInputSnapshot.java @@ -1,5 +1,6 @@ package it.cnr.isti.workflow.manager.executions.persistence; +import it.cnr.isti.workflow.manager.executions.steps.InputResolutionState; import lombok.AllArgsConstructor; import lombok.Builder; import lombok.Data; @@ -15,4 +16,5 @@ public class ExecutionInputSnapshot { private Object value; private boolean registered; private boolean unavailable; + private InputResolutionState resolutionState; } diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionSnapshot.java b/src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionSnapshot.java index ba95965..353825d 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionSnapshot.java +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionSnapshot.java @@ -20,6 +20,7 @@ import lombok.NoArgsConstructor; @AllArgsConstructor public class ExecutionSnapshot { + private Integer snapshotVersion; private ExecutionStatus status; private Long startTime; private Long endTime; diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionStepSnapshot.java b/src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionStepSnapshot.java index e246a5d..c67d8b0 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionStepSnapshot.java +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionStepSnapshot.java @@ -4,6 +4,7 @@ import java.util.List; import java.util.Map; import it.cnr.isti.workflow.manager.executions.steps.StepStatus; +import it.cnr.isti.workflow.manager.executions.steps.StepSkipReason; import lombok.AllArgsConstructor; import lombok.Builder; import lombok.Data; @@ -21,4 +22,5 @@ public class ExecutionStepSnapshot { private boolean simulated; private List inputs; private Map partialResults; + private StepSkipReason skipReason; } diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/steps/Input.java b/src/main/java/it/cnr/isti/workflow/manager/executions/steps/Input.java index f8ccfb8..0f0033b 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/executions/steps/Input.java +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/steps/Input.java @@ -19,7 +19,7 @@ public class Input { private Object value = null; private boolean registered = false; - private boolean unavailable = false; + private InputResolutionState resolutionState = InputResolutionState.PENDING; @JsonIgnore private InputListener listener; @@ -37,17 +37,27 @@ public class Input { public void setValue(Object value) { validateValue(value); this.value = value; - this.unavailable = false; + this.resolutionState = value == null ? InputResolutionState.PENDING : InputResolutionState.VALUE; if (listener != null) { listener.onInputSet(); } } public void markUnavailable() { - if (this.value != null || this.unavailable) { + if (this.resolutionState != InputResolutionState.PENDING) { return; } - this.unavailable = true; + this.resolutionState = InputResolutionState.UNAVAILABLE; + if (listener != null) { + listener.onInputUnavailable(); + } + } + + public void markNotSelected() { + if (this.resolutionState != InputResolutionState.PENDING) { + return; + } + this.resolutionState = InputResolutionState.NOT_SELECTED; if (listener != null) { listener.onInputUnavailable(); } @@ -58,15 +68,20 @@ public class Input { } public boolean isSet() { - return this.value != null; + return this.resolutionState == InputResolutionState.VALUE; } public boolean isUnavailable() { - return this.unavailable; + return this.resolutionState == InputResolutionState.UNAVAILABLE + || this.resolutionState == InputResolutionState.NOT_SELECTED; + } + + public boolean isNotSelected() { + return this.resolutionState == InputResolutionState.NOT_SELECTED; } public boolean isSatisfied() { - return isSet() || isUnavailable(); + return this.resolutionState != InputResolutionState.PENDING; } protected void setListener(InputListener listener) { @@ -78,14 +93,30 @@ public class Input { .name(descriptor.getName()) .value(value) .registered(registered) - .unavailable(unavailable) + .unavailable(isUnavailable()) + .resolutionState(resolutionState) .build(); } protected void restore(ExecutionInputSnapshot snapshot) { this.value = normalizeRestoredValue(snapshot == null ? null : snapshot.getValue()); this.registered = snapshot != null && snapshot.isRegistered(); - this.unavailable = snapshot != null && snapshot.isUnavailable(); + this.resolutionState = resolveRestoredState(snapshot); + } + + private InputResolutionState resolveRestoredState(ExecutionInputSnapshot snapshot) { + if (snapshot == null) { + return InputResolutionState.PENDING; + } + if (snapshot.getResolutionState() != null) { + return snapshot.getResolutionState(); + } + if (snapshot.getValue() != null) { + return InputResolutionState.VALUE; + } + return snapshot.isUnavailable() + ? InputResolutionState.UNAVAILABLE + : InputResolutionState.PENDING; } private void validateValue(Object value) { diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/steps/InputResolutionState.java b/src/main/java/it/cnr/isti/workflow/manager/executions/steps/InputResolutionState.java new file mode 100644 index 0000000..104d867 --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/steps/InputResolutionState.java @@ -0,0 +1,8 @@ +package it.cnr.isti.workflow.manager.executions.steps; + +public enum InputResolutionState { + PENDING, + VALUE, + NOT_SELECTED, + UNAVAILABLE +} diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/steps/Output.java b/src/main/java/it/cnr/isti/workflow/manager/executions/steps/Output.java index 19897c7..702090b 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/executions/steps/Output.java +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/steps/Output.java @@ -34,6 +34,12 @@ public class Output { } } + public void markNotSelected() { + for (Input source : sources) { + source.markNotSelected(); + } + } + public boolean isConnected() { return !sources.isEmpty(); } diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/steps/Step.java b/src/main/java/it/cnr/isti/workflow/manager/executions/steps/Step.java index 78e08af..1a5ef98 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/executions/steps/Step.java +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/steps/Step.java @@ -13,11 +13,15 @@ import com.fasterxml.jackson.annotation.JsonIgnore; import it.cnr.isti.workflow.manager.blocks.Block; import it.cnr.isti.workflow.manager.executions.ExecutionEventLogger; import it.cnr.isti.workflow.manager.executions.ExecutionListener; +import it.cnr.isti.workflow.manager.executions.NodeExecutionResult; import it.cnr.isti.workflow.manager.executions.persistence.ExecutionInputSnapshot; import it.cnr.isti.workflow.manager.executions.persistence.ExecutionStepSnapshot; import it.cnr.isti.workflow.manager.executions.executors.NodeExecutors; import it.cnr.isti.workflow.manager.executions.bias.BiasExecutionContext; import it.cnr.isti.workflow.manager.executions.bias.BiasApiException; +import it.cnr.isti.workflow.manager.executions.steps.activation.StepActivationDecision; +import it.cnr.isti.workflow.manager.executions.steps.activation.StepActivationPolicy; +import it.cnr.isti.workflow.manager.executions.steps.activation.StepActivationPolicyRegistry; import it.cnr.isti.workflow.manager.flows.model.FlowNode; import it.cnr.isti.workflow.manager.llms.LLMDescriptor; import lombok.Builder; @@ -58,7 +62,7 @@ public class Step implements InputListener { private final Set satisfiedDependencyIds = new LinkedHashSet<>(); @JsonIgnore - private boolean dependencyUnavailable = false; + private StepSkipReason dependencySkipReason; @Getter @JsonIgnore @@ -67,6 +71,9 @@ public class Step implements InputListener { @Getter private StepStatus status = StepStatus.WAITING_FOR_INPUT; + @Getter + private StepSkipReason skipReason; + @Getter @JsonIgnore private boolean started = false; @@ -111,11 +118,15 @@ public class Step implements InputListener { @JsonIgnore private ExecutionEventLogger eventLogger; + @JsonIgnore + private StepActivationPolicy activationPolicy; + @Builder public Step(@NonNull N node) { // Initialize the step with the provided block this.node = node; this.id = node.getId(); + this.activationPolicy = StepActivationPolicyRegistry.resolve(node); this.node.getOutputs().forEach(outputName -> { Output output = new Output(outputName); this.outputs.add(output); @@ -175,15 +186,18 @@ public class Step implements InputListener { this.status = StepStatus.RUNNING; listener.started(this.id); try { - Map outputs = isSimulated() && this.node.isUserInteractive() - ? NodeExecutors.simulate(this.node, this.inputs, authorizations, executionVariables, + NodeExecutionResult executionResult = isSimulated() && this.node.isUserInteractive() + ? NodeExecutors.simulateResult(this.node, this.inputs, authorizations, executionVariables, executionVariableDescriptors, this.interactionSimulationDescriptor, this.eventLogger, this.biasExecutionContext) - : NodeExecutors.execute(this.node, this.inputs, authorizations, executionVariables, + : NodeExecutors.executeResult(this.node, this.inputs, authorizations, executionVariables, executionVariableDescriptors, this.eventLogger, this.biasExecutionContext); + Map outputs = executionResult.outputs(); for (Output output : this.outputs) { if (outputs.containsKey(output.getDescriptor().getName())) { output.setValue(outputs.get(output.getDescriptor().getName())); + } else if (executionResult.notSelectedOutputs().contains(output.getDescriptor().getName())) { + output.markNotSelected(); } } this.status = StepStatus.COMPLETED; @@ -257,15 +271,21 @@ public class Step implements InputListener { } public synchronized void dependencySkipped(String stepId) { + dependencySkipped(stepId, StepSkipReason.DEPENDENCY_UNAVAILABLE); + } + + public synchronized void dependencySkipped(String stepId, StepSkipReason upstreamSkipReason) { if (this.dependencyIds.contains(stepId)) { - this.dependencyUnavailable = true; + this.dependencySkipReason = upstreamSkipReason == StepSkipReason.BRANCH_NOT_SELECTED + ? StepSkipReason.BRANCH_NOT_SELECTED + : StepSkipReason.DEPENDENCY_UNAVAILABLE; refreshStateFromInputs(); } } public synchronized void resetDependencyState() { this.satisfiedDependencyIds.clear(); - this.dependencyUnavailable = false; + this.dependencySkipReason = null; refreshStateFromInputs(); } @@ -286,25 +306,27 @@ public class Step implements InputListener { } private void refreshStateFromInputs() { - if (this.inputs.stream().anyMatch(Input::isUnavailable) || this.dependencyUnavailable) { - skip(); - return; - } if (this.status == StepStatus.WAITING_FOR_INTERACTION) { return; } - if (!inputs.stream().allMatch(Input::isSet)) { - this.status = StepStatus.WAITING_FOR_INPUT; - return; + if (this.activationPolicy == null) { + this.activationPolicy = StepActivationPolicyRegistry.resolve(this.node); } - if (areDependenciesSatisfied()) { - this.status = StepStatus.READY; - if (this.started) { - scheduleRun(); + StepActivationDecision decision = this.activationPolicy.evaluate( + this.inputs, areDependenciesSatisfied(), this.dependencySkipReason); + switch (decision) { + case SKIP_BRANCH_NOT_SELECTED -> skip(StepSkipReason.BRANCH_NOT_SELECTED); + case SKIP_INPUT_UNAVAILABLE -> skip(StepSkipReason.INPUT_UNAVAILABLE); + case SKIP_DEPENDENCY_UNAVAILABLE -> skip(StepSkipReason.DEPENDENCY_UNAVAILABLE); + case WAITING_FOR_INPUT -> this.status = StepStatus.WAITING_FOR_INPUT; + case WAITING_FOR_DEPENDENCY -> this.status = StepStatus.WAITING_FOR_DEPENDENCY; + case READY -> { + this.status = StepStatus.READY; + if (this.started) { + scheduleRun(); + } } - return; } - this.status = StepStatus.WAITING_FOR_DEPENDENCY; } private void scheduleRun() { @@ -318,14 +340,15 @@ public class Step implements InputListener { }); } - private void skip() { + private void skip(StepSkipReason reason) { if (isTerminal()) { return; } + this.skipReason = reason; this.status = StepStatus.SKIPPED; logger.info("Skipping step " + this.id + " of node " + this.node.getName()); if (this.listener != null) { - listener.skipped(this.id); + listener.skipped(this.id, reason); } } @@ -382,6 +405,7 @@ public class Step implements InputListener { .simulated(this.simulated) .inputs(this.inputs.stream().map(Input::snapshot).toList()) .partialResults(Map.copyOf(this.partialResults)) + .skipReason(this.skipReason) .build(); } @@ -402,6 +426,7 @@ public class Step implements InputListener { } this.started = snapshot.isStarted(); this.simulated = snapshot.isSimulated(); + this.skipReason = snapshot.getSkipReason(); this.status = normalizeRestoredStatus(snapshot.getStatus()); } diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/steps/StepSkipReason.java b/src/main/java/it/cnr/isti/workflow/manager/executions/steps/StepSkipReason.java new file mode 100644 index 0000000..27a6e88 --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/steps/StepSkipReason.java @@ -0,0 +1,7 @@ +package it.cnr.isti.workflow.manager.executions.steps; + +public enum StepSkipReason { + BRANCH_NOT_SELECTED, + INPUT_UNAVAILABLE, + DEPENDENCY_UNAVAILABLE +} diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/steps/activation/AllRequiredActivationPolicy.java b/src/main/java/it/cnr/isti/workflow/manager/executions/steps/activation/AllRequiredActivationPolicy.java new file mode 100644 index 0000000..80b93e4 --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/steps/activation/AllRequiredActivationPolicy.java @@ -0,0 +1,32 @@ +package it.cnr.isti.workflow.manager.executions.steps.activation; + +import java.util.List; + +import it.cnr.isti.workflow.manager.executions.steps.Input; +import it.cnr.isti.workflow.manager.executions.steps.InputResolutionState; +import it.cnr.isti.workflow.manager.executions.steps.StepSkipReason; + +public final class AllRequiredActivationPolicy implements StepActivationPolicy { + + @Override + public StepActivationDecision evaluate(List inputs, boolean dependenciesSatisfied, + StepSkipReason dependencySkipReason) { + if (inputs.stream().anyMatch(input -> input.getResolutionState() == InputResolutionState.NOT_SELECTED)) { + return StepActivationDecision.SKIP_BRANCH_NOT_SELECTED; + } + if (inputs.stream().anyMatch(input -> input.getResolutionState() == InputResolutionState.UNAVAILABLE)) { + return StepActivationDecision.SKIP_INPUT_UNAVAILABLE; + } + if (dependencySkipReason != null) { + return dependencySkipReason == StepSkipReason.BRANCH_NOT_SELECTED + ? StepActivationDecision.SKIP_BRANCH_NOT_SELECTED + : StepActivationDecision.SKIP_DEPENDENCY_UNAVAILABLE; + } + if (inputs.stream().anyMatch(input -> input.getResolutionState() != InputResolutionState.VALUE)) { + return StepActivationDecision.WAITING_FOR_INPUT; + } + return dependenciesSatisfied + ? StepActivationDecision.READY + : StepActivationDecision.WAITING_FOR_DEPENDENCY; + } +} diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/steps/activation/StepActivationDecision.java b/src/main/java/it/cnr/isti/workflow/manager/executions/steps/activation/StepActivationDecision.java new file mode 100644 index 0000000..8ce420e --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/steps/activation/StepActivationDecision.java @@ -0,0 +1,10 @@ +package it.cnr.isti.workflow.manager.executions.steps.activation; + +public enum StepActivationDecision { + WAITING_FOR_INPUT, + WAITING_FOR_DEPENDENCY, + READY, + SKIP_BRANCH_NOT_SELECTED, + SKIP_INPUT_UNAVAILABLE, + SKIP_DEPENDENCY_UNAVAILABLE +} diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/steps/activation/StepActivationPolicy.java b/src/main/java/it/cnr/isti/workflow/manager/executions/steps/activation/StepActivationPolicy.java new file mode 100644 index 0000000..5985eef --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/steps/activation/StepActivationPolicy.java @@ -0,0 +1,12 @@ +package it.cnr.isti.workflow.manager.executions.steps.activation; + +import java.util.List; + +import it.cnr.isti.workflow.manager.executions.steps.Input; +import it.cnr.isti.workflow.manager.executions.steps.StepSkipReason; + +public interface StepActivationPolicy { + + StepActivationDecision evaluate(List inputs, boolean dependenciesSatisfied, + StepSkipReason dependencySkipReason); +} diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/steps/activation/StepActivationPolicyRegistry.java b/src/main/java/it/cnr/isti/workflow/manager/executions/steps/activation/StepActivationPolicyRegistry.java new file mode 100644 index 0000000..65a9ff0 --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/steps/activation/StepActivationPolicyRegistry.java @@ -0,0 +1,18 @@ +package it.cnr.isti.workflow.manager.executions.steps.activation; + +import it.cnr.isti.workflow.manager.flows.model.FlowNode; + +public final class StepActivationPolicyRegistry { + + private static final StepActivationPolicy DEFAULT_POLICY = new AllRequiredActivationPolicy(); + + private StepActivationPolicyRegistry() { + } + + public static StepActivationPolicy resolve(FlowNode node) { + if (node == null) { + throw new IllegalArgumentException("Flow node is required to resolve its activation policy"); + } + return DEFAULT_POLICY; + } +} diff --git a/src/test/java/it/cnr/isti/workflow/manager/executions/ExecutionTest.java b/src/test/java/it/cnr/isti/workflow/manager/executions/ExecutionTest.java index 6bc083e..a77faa7 100644 --- a/src/test/java/it/cnr/isti/workflow/manager/executions/ExecutionTest.java +++ b/src/test/java/it/cnr/isti/workflow/manager/executions/ExecutionTest.java @@ -40,6 +40,8 @@ import it.cnr.isti.workflow.manager.containers.configurations.IteratorContainerC import it.cnr.isti.workflow.manager.containers.configurations.LoopContainerConfiguration; import it.cnr.isti.workflow.manager.executions.steps.Step; import it.cnr.isti.workflow.manager.executions.steps.StepStatus; +import it.cnr.isti.workflow.manager.executions.steps.InputResolutionState; +import it.cnr.isti.workflow.manager.executions.steps.StepSkipReason; import it.cnr.isti.workflow.manager.containers.factories.GenericContainerFactory; import it.cnr.isti.workflow.manager.containers.factories.IteratorContainerFactory; import it.cnr.isti.workflow.manager.containers.factories.LoopContainerFactory; @@ -854,6 +856,78 @@ public class ExecutionTest { assertFalse(execObject.getContext().getResult().containsKey(new FieldKey(switchBlock.getId(), "rejected"))); } + @Test + public void switchExecutionSkipsUnselectedBranchAndItsDownstream() { + Block switchBlock = switchBlockFactory.create(SwitchBlockConfiguration.builder() + .name("Decision") + .cases(List.of(new SwitchCase("approved"), new SwitchCase("rejected"))) + .condition("${{decision}}") + .useLlm(false) + .outputTemplate("${{candidate}}") + .build()); + Block approved = llmBlockFactory.create(LLMBlockConfiguration.builder() + .name("Approved branch") + .prompt("Approve ${{candidate}}") + .llmDescriptor(llmBrick) + .build()); + Block rejected = llmBlockFactory.create(LLMBlockConfiguration.builder() + .name("Rejected branch") + .prompt("Reject ${{candidate}}") + .llmDescriptor(llmBrick) + .build()); + Block rejectedDownstream = llmBlockFactory.create(LLMBlockConfiguration.builder() + .name("Rejected downstream") + .prompt("Archive ${{rejection}}") + .llmDescriptor(llmBrick) + .build()); + + FlowData flow = FlowData.builder() + .block(switchBlock) + .block(approved) + .block(rejected) + .block(rejectedDownstream) + .connection(Connection.builder() + .sourceId(switchBlock.getId()) + .sourceName("approved") + .targetId(approved.getId()) + .targetName("candidate") + .build()) + .connection(Connection.builder() + .sourceId(switchBlock.getId()) + .sourceName("rejected") + .targetId(rejected.getId()) + .targetName("candidate") + .build()) + .connection(Connection.builder() + .sourceId(rejected.getId()) + .sourceName(LLMBlockFactory.OUTPUT_NAME) + .targetId(rejectedDownstream.getId()) + .targetName("rejection") + .build()) + .build(); + + ExecutionObject execution = executionsService.createExecution("Characterize exclusive routing", flow); + executionsService.prepareInput(execution.getId(), switchBlock.getId(), "decision", "approved"); + executionsService.prepareInput(execution.getId(), switchBlock.getId(), "candidate", "Ada"); + execution = executionsService.startExecution(execution.getId()); + while (execution.getContext().getStatus() == ExecutionStatus.RUNNING) { + execution = executionsService.getExecution(execution.getId()); + } + + assertEquals(ExecutionStatus.SUCCESS, execution.getContext().getStatus()); + assertEquals(StepStatus.COMPLETED, execution.getContext().getSteps().get(switchBlock.getId()).getStatus()); + assertEquals(StepStatus.COMPLETED, execution.getContext().getSteps().get(approved.getId()).getStatus()); + assertEquals(StepStatus.SKIPPED, execution.getContext().getSteps().get(rejected.getId()).getStatus()); + assertEquals(StepStatus.SKIPPED, execution.getContext().getSteps().get(rejectedDownstream.getId()).getStatus()); + assertEquals(StepSkipReason.BRANCH_NOT_SELECTED, + execution.getContext().getSteps().get(rejected.getId()).getSkipReason()); + assertEquals(StepSkipReason.BRANCH_NOT_SELECTED, + execution.getContext().getSteps().get(rejectedDownstream.getId()).getSkipReason()); + assertEquals(InputResolutionState.NOT_SELECTED, + execution.getContext().getSteps().get(rejected.getId()).getInputs().getFirst().getResolutionState()); + assertEquals(2, execution.snapshot().getSnapshotVersion()); + } + @Test public void switchExecutionRoutesToConfiguredOutputUsingLlm() { Block switchBlock = switchBlockFactory.create(SwitchBlockConfiguration.builder() diff --git a/src/test/java/it/cnr/isti/workflow/manager/executions/NodeExecutionResultTest.java b/src/test/java/it/cnr/isti/workflow/manager/executions/NodeExecutionResultTest.java new file mode 100644 index 0000000..0bed525 --- /dev/null +++ b/src/test/java/it/cnr/isti/workflow/manager/executions/NodeExecutionResultTest.java @@ -0,0 +1,27 @@ +package it.cnr.isti.workflow.manager.executions; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.util.Map; +import java.util.Set; + +import org.junit.jupiter.api.Test; + +class NodeExecutionResultTest { + + @Test + void routedResultMarksEveryUnproducedBranchAsNotSelected() { + NodeExecutionResult result = NodeExecutionResult.routed( + Map.of("approved", "payload"), Set.of("approved", "rejected", "manual-review")); + + assertEquals(Map.of("approved", "payload"), result.outputs()); + assertEquals(Set.of("rejected", "manual-review"), result.notSelectedOutputs()); + } + + @Test + void outputCannotAlsoBeDeclaredNotSelected() { + assertThrows(IllegalArgumentException.class, + () -> new NodeExecutionResult(Map.of("approved", "payload"), Set.of("approved"))); + } +} diff --git a/src/test/java/it/cnr/isti/workflow/manager/executions/steps/InputResolutionAndActivationPolicyTest.java b/src/test/java/it/cnr/isti/workflow/manager/executions/steps/InputResolutionAndActivationPolicyTest.java new file mode 100644 index 0000000..043090d --- /dev/null +++ b/src/test/java/it/cnr/isti/workflow/manager/executions/steps/InputResolutionAndActivationPolicyTest.java @@ -0,0 +1,102 @@ +package it.cnr.isti.workflow.manager.executions.steps; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.List; + +import org.junit.jupiter.api.Test; + +import it.cnr.isti.workflow.manager.executions.persistence.ExecutionInputSnapshot; +import it.cnr.isti.workflow.manager.executions.steps.activation.AllRequiredActivationPolicy; +import it.cnr.isti.workflow.manager.executions.steps.activation.StepActivationDecision; +import it.cnr.isti.workflow.manager.ios.IODescriptor; +import it.cnr.isti.workflow.manager.ios.IOType; + +class InputResolutionAndActivationPolicyTest { + + private final IODescriptor textInput = IODescriptor.of("input", IOType.TEXT); + private final AllRequiredActivationPolicy policy = new AllRequiredActivationPolicy(); + + @Test + void inputTracksValueNotSelectedAndUnavailableAsDistinctStates() { + Input value = new Input(textInput); + assertEquals(InputResolutionState.PENDING, value.getResolutionState()); + value.setValue("payload"); + assertEquals(InputResolutionState.VALUE, value.getResolutionState()); + assertTrue(value.isSet()); + assertFalse(value.isUnavailable()); + + Input notSelected = new Input(textInput); + notSelected.markNotSelected(); + assertEquals(InputResolutionState.NOT_SELECTED, notSelected.getResolutionState()); + assertTrue(notSelected.isNotSelected()); + assertTrue(notSelected.isUnavailable()); + + Input unavailable = new Input(textInput); + unavailable.markUnavailable(); + assertEquals(InputResolutionState.UNAVAILABLE, unavailable.getResolutionState()); + assertFalse(unavailable.isNotSelected()); + assertTrue(unavailable.isUnavailable()); + } + + @Test + void snapshotsPersistNewStateAndRestoreLegacyState() { + Input notSelected = new Input(textInput); + notSelected.markNotSelected(); + ExecutionInputSnapshot currentSnapshot = notSelected.snapshot(); + assertEquals(InputResolutionState.NOT_SELECTED, currentSnapshot.getResolutionState()); + + Input currentRestored = new Input(textInput); + currentRestored.restore(currentSnapshot); + assertEquals(InputResolutionState.NOT_SELECTED, currentRestored.getResolutionState()); + + ExecutionInputSnapshot legacyValue = ExecutionInputSnapshot.builder() + .name("input") + .value("legacy") + .registered(true) + .unavailable(false) + .build(); + Input restoredValue = new Input(textInput); + restoredValue.restore(legacyValue); + assertEquals(InputResolutionState.VALUE, restoredValue.getResolutionState()); + assertEquals("legacy", restoredValue.getValue()); + + ExecutionInputSnapshot legacyUnavailable = ExecutionInputSnapshot.builder() + .name("input") + .registered(true) + .unavailable(true) + .build(); + Input restoredUnavailable = new Input(textInput); + restoredUnavailable.restore(legacyUnavailable); + assertEquals(InputResolutionState.UNAVAILABLE, restoredUnavailable.getResolutionState()); + } + + @Test + void allRequiredPolicyPreservesLegacyReadinessAndAddsSkipCause() { + Input value = new Input(textInput); + value.setValue("payload"); + assertEquals(StepActivationDecision.READY, policy.evaluate(List.of(value), true, null)); + assertEquals(StepActivationDecision.WAITING_FOR_DEPENDENCY, + policy.evaluate(List.of(value), false, null)); + + Input pending = new Input(textInput); + assertEquals(StepActivationDecision.WAITING_FOR_INPUT, + policy.evaluate(List.of(pending), true, null)); + + Input notSelected = new Input(textInput); + notSelected.markNotSelected(); + assertEquals(StepActivationDecision.SKIP_BRANCH_NOT_SELECTED, + policy.evaluate(List.of(notSelected), true, null)); + + Input unavailable = new Input(textInput); + unavailable.markUnavailable(); + assertEquals(StepActivationDecision.SKIP_INPUT_UNAVAILABLE, + policy.evaluate(List.of(unavailable), true, null)); + assertEquals(StepActivationDecision.SKIP_BRANCH_NOT_SELECTED, + policy.evaluate(List.of(), true, StepSkipReason.BRANCH_NOT_SELECTED)); + assertEquals(StepActivationDecision.SKIP_DEPENDENCY_UNAVAILABLE, + policy.evaluate(List.of(), true, StepSkipReason.DEPENDENCY_UNAVAILABLE)); + } +}