lelii
/
humainflow-service
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: add control-flow runtime foundations

This commit is contained in:
Lucio Lelii 2026-07-22 10:49:21 +02:00
parent 3280584d31
commit b8fbbe8de0
20 changed files with 504 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionContext.java
View File

@ -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<String, Object> providedAuthorizations) {
return ExecutionSnapshot.builder()
.snapshotVersion(2)
.status(this.status)
.startTime(this.startTime)
.endTime(this.endTime)

6
src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionListener.java
View File

@ -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<String, Object> 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) {

3
src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionObject.java
View File

@ -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 -> {
}
}

30
src/main/java/it/cnr/isti/workflow/manager/executions/NodeExecutionResult.java Normal file
View File

@ -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<String, Object> outputs,
Set<String> notSelectedOutputs) {
public NodeExecutionResult {
outputs = outputs == null ? Map.of() : Map.copyOf(outputs);
notSelectedOutputs = notSelectedOutputs == null ? Set.of() : Set.copyOf(notSelectedOutputs);
Set<String> 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<String, Object> outputs) {
return new NodeExecutionResult(outputs, Set.of());
}
public static NodeExecutionResult routed(Map<String, Object> outputs, Set<String> declaredOutputs) {
Set<String> notSelected = new LinkedHashSet<>(declaredOutputs == null ? Set.of() : declaredOutputs);
notSelected.removeAll(outputs == null ? Set.of() : outputs.keySet());
return new NodeExecutionResult(outputs, notSelected);
}
}

75
src/main/java/it/cnr/isti/workflow/manager/executions/executors/NodeExecutors.java
View File

@ -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<String, Object> execute(FlowNode node, List<Input> inputs, Map<String, Object> authorizations,
Map<String, Object> executionVariables, Map<String, ExecutionVariableDescriptor> 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<String, Object> execute(FlowNode node, List<Input> inputs, Map<String, Object> authorizations,
Map<String, Object> executionVariables, Map<String, ExecutionVariableDescriptor> executionVariableDescriptors,
ExecutionEventLogger eventLogger, BiasExecutionContext biasExecutionContext) {
return executeResult(node, inputs, authorizations, executionVariables, executionVariableDescriptors, eventLogger,
biasExecutionContext).outputs();
}
public static NodeExecutionResult executeResult(FlowNode node, List<Input> inputs,
Map<String, Object> authorizations, Map<String, Object> executionVariables,
Map<String, ExecutionVariableDescriptor> executionVariableDescriptors,
ExecutionEventLogger eventLogger) {
return executeResult(node, inputs, authorizations, executionVariables, executionVariableDescriptors, eventLogger,
BiasExecutionContext.normal());
}
public static NodeExecutionResult executeResult(FlowNode node, List<Input> inputs,
Map<String, Object> authorizations, Map<String, Object> executionVariables,
Map<String, ExecutionVariableDescriptor> executionVariableDescriptors,
ExecutionEventLogger eventLogger, BiasExecutionContext biasExecutionContext) {
Map<String, Object> 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<String, Object> simulate(FlowNode node, List<Input> inputs, Map<String, Object> authorizations,
Map<String, Object> executionVariables, Map<String, ExecutionVariableDescriptor> 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<String, Object> simulate(FlowNode node, List<Input> inputs, Map<String, Object> authorizations,
Map<String, Object> executionVariables, Map<String, ExecutionVariableDescriptor> 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<Input> inputs,
Map<String, Object> authorizations, Map<String, Object> executionVariables,
Map<String, ExecutionVariableDescriptor> executionVariableDescriptors,
LLMDescriptor simulatorDescriptor, ExecutionEventLogger eventLogger,
BiasExecutionContext biasExecutionContext) {
if (node instanceof Block<?> block) {
return simulateBlock(block, inputs, authorizations, executionVariables, executionVariableDescriptors,
simulatorDescriptor, eventLogger, biasExecutionContext);
Map<String, Object> 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<Input> inputs,
Map<String, Object> authorizations, Map<String, Object> executionVariables,
Map<String, ExecutionVariableDescriptor> executionVariableDescriptors,
LLMDescriptor simulatorDescriptor, ExecutionEventLogger eventLogger) {
return simulateResult(node, inputs, authorizations, executionVariables, executionVariableDescriptors,
simulatorDescriptor, eventLogger, BiasExecutionContext.normal());
}
public static InteractionResult interact(FlowNode node, List<Input> inputs, Map<String, Object> interaction,
Map<String, Object> partialResults, Map<String, Object> authorizations,
Map<String, Object> executionVariables, Map<String, ExecutionVariableDescriptor> 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<String, Object> outputs) {
if (node instanceof Block<?> block
&& (block.getSpecificConfiguration() instanceof ConditionalBlockConfiguration
|| block.getSpecificConfiguration() instanceof SwitchBlockConfiguration)) {
Set<String> declaredOutputs = node.getOutputs().stream()
.map(output -> output.getName())
.collect(Collectors.toCollection(java.util.LinkedHashSet::new));
return NodeExecutionResult.routed(outputs, declaredOutputs);
}
return NodeExecutionResult.completed(outputs);
}
}

2
src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionInputSnapshot.java
View File

@ -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;
}

1
src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionSnapshot.java
View File

@ -20,6 +20,7 @@ import lombok.NoArgsConstructor;
@AllArgsConstructor
public class ExecutionSnapshot {
private Integer snapshotVersion;
private ExecutionStatus status;
private Long startTime;
private Long endTime;

2
src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionStepSnapshot.java
View File

@ -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<ExecutionInputSnapshot> inputs;
private Map<String, Object> partialResults;
private StepSkipReason skipReason;
}

49
src/main/java/it/cnr/isti/workflow/manager/executions/steps/Input.java
View File

@ -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) {

8
src/main/java/it/cnr/isti/workflow/manager/executions/steps/InputResolutionState.java Normal file
View File

@ -0,0 +1,8 @@
package it.cnr.isti.workflow.manager.executions.steps;
public enum InputResolutionState {
PENDING,
VALUE,
NOT_SELECTED,
UNAVAILABLE
}

6
src/main/java/it/cnr/isti/workflow/manager/executions/steps/Output.java
View File

@ -34,6 +34,12 @@ public class Output {
}
}
public void markNotSelected() {
for (Input source : sources) {
source.markNotSelected();
}
}
public boolean isConnected() {
return !sources.isEmpty();
}

67
src/main/java/it/cnr/isti/workflow/manager/executions/steps/Step.java
View File

@ -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<N extends FlowNode> implements InputListener {
private final Set<String> satisfiedDependencyIds = new LinkedHashSet<>();
@JsonIgnore
private boolean dependencyUnavailable = false;
private StepSkipReason dependencySkipReason;
@Getter
@JsonIgnore
@ -67,6 +71,9 @@ public class Step<N extends FlowNode> 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<N extends FlowNode> 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<N extends FlowNode> implements InputListener {
this.status = StepStatus.RUNNING;
listener.started(this.id);
try {
Map<String, Object> 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<String, Object> 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<N extends FlowNode> 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<N extends FlowNode> 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<N extends FlowNode> 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<N extends FlowNode> 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<N extends FlowNode> implements InputListener {
}
this.started = snapshot.isStarted();
this.simulated = snapshot.isSimulated();
this.skipReason = snapshot.getSkipReason();
this.status = normalizeRestoredStatus(snapshot.getStatus());
}

7
src/main/java/it/cnr/isti/workflow/manager/executions/steps/StepSkipReason.java Normal file
View File

@ -0,0 +1,7 @@
package it.cnr.isti.workflow.manager.executions.steps;
public enum StepSkipReason {
BRANCH_NOT_SELECTED,
INPUT_UNAVAILABLE,
DEPENDENCY_UNAVAILABLE
}

32
src/main/java/it/cnr/isti/workflow/manager/executions/steps/activation/AllRequiredActivationPolicy.java Normal file
View File

@ -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<Input> 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;
}
}

10
src/main/java/it/cnr/isti/workflow/manager/executions/steps/activation/StepActivationDecision.java Normal file
View File

@ -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
}

12
src/main/java/it/cnr/isti/workflow/manager/executions/steps/activation/StepActivationPolicy.java Normal file
View File

@ -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<Input> inputs, boolean dependenciesSatisfied,
StepSkipReason dependencySkipReason);
}

18
src/main/java/it/cnr/isti/workflow/manager/executions/steps/activation/StepActivationPolicyRegistry.java Normal file
View File

@ -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;
}
}

74
src/test/java/it/cnr/isti/workflow/manager/executions/ExecutionTest.java
View File

@ -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<SwitchBlockType> switchBlock = switchBlockFactory.create(SwitchBlockConfiguration.builder()
.name("Decision")
.cases(List.of(new SwitchCase("approved"), new SwitchCase("rejected")))
.condition("${{decision}}")
.useLlm(false)
.outputTemplate("${{candidate}}")
.build());
Block<LLMBlockType> approved = llmBlockFactory.create(LLMBlockConfiguration.builder()
.name("Approved branch")
.prompt("Approve ${{candidate}}")
.llmDescriptor(llmBrick)
.build());
Block<LLMBlockType> rejected = llmBlockFactory.create(LLMBlockConfiguration.builder()
.name("Rejected branch")
.prompt("Reject ${{candidate}}")
.llmDescriptor(llmBrick)
.build());
Block<LLMBlockType> 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<SwitchBlockType> switchBlock = switchBlockFactory.create(SwitchBlockConfiguration.builder()

27
src/test/java/it/cnr/isti/workflow/manager/executions/NodeExecutionResultTest.java Normal file
View File

@ -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")));
}
}

102
src/test/java/it/cnr/isti/workflow/manager/executions/steps/InputResolutionAndActivationPolicyTest.java Normal file
View File

@ -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));
}
}