feat: suspend generic containers for subflows

This commit is contained in:
Lucio Lelii 2026-07-23 19:25:21 +02:00
parent a64642923f
commit 0e5831289f
20 changed files with 335 additions and 70 deletions

View File

@ -54,15 +54,6 @@ public class ContainerSubFlowValidator {
"type",
"Nested containers are not supported")));
subFlow.getNodes().stream()
.filter(node -> node != null && node.isUserInteractive())
.findFirst()
.ifPresent(node -> errors.add(new ValidationError(
ValidationErrorCode.INTERACTIVE_NODES_IN_CONTAINER_NOT_SUPPORTED,
"node",
node.getId(),
"type",
"Interactive nodes inside containers are not supported yet")));
}
errors.addAll(validationRegistry.validate(validationType, subFlow, openInputs, openOutputs));

View File

@ -0,0 +1,14 @@
package it.cnr.isti.workflow.manager.executions;
/** Stable identity of a parent execution and its active container step. */
public record ContainerExecutionContext(String parentExecutionId, String parentStepId) {
public ContainerExecutionContext {
if (parentExecutionId == null || parentExecutionId.isBlank()) {
throw new IllegalArgumentException("Parent execution id is required");
}
if (parentStepId == null || parentStepId.isBlank()) {
throw new IllegalArgumentException("Parent step id is required");
}
}
}

View File

@ -303,8 +303,15 @@ public class ExecutionContext implements ExecutionListener {
return;
}
logger.info("Step " + id + " paused");
addStepEvent(id, ExecutionEventLevel.INFO, ExecutionEventType.STEP_WAITING_FOR_INTERACTION,
"Waiting for interaction on " + resolveNodeName(id), Map.of());
Step<?> step = this.steps.get(id);
boolean waitingForSubflow = step != null && step.getStatus() == StepStatus.WAITING_FOR_SUBFLOW;
addStepEvent(id, ExecutionEventLevel.INFO,
waitingForSubflow ? ExecutionEventType.STEP_WAITING_FOR_SUBFLOW : ExecutionEventType.STEP_WAITING_FOR_INTERACTION,
waitingForSubflow ? "Waiting for subflow on " + resolveNodeName(id)
: "Waiting for interaction on " + resolveNodeName(id),
waitingForSubflow && step.getContainerContinuation() != null
? Map.of("innerExecutionId", step.getContainerContinuation().getActiveInnerExecutionId())
: Map.of());
synchronized (this.waitingSteps) {
if (!this.waitingSteps.contains(id)) {
this.waitingSteps.add(id);
@ -391,6 +398,24 @@ public class ExecutionContext implements ExecutionListener {
notifyStateChanged();
}
protected void completeContainerSubflow(String stepId, NodeExecutionResult result) {
Step<?> step = this.steps.get(stepId);
if (step == null) {
throw new IllegalArgumentException("Step with id " + stepId + " not found");
}
step.completeSuspendedContainer(result);
notifyStateChanged();
}
protected void failContainerSubflow(String stepId, String error) {
Step<?> step = this.steps.get(stepId);
if (step == null) {
throw new IllegalArgumentException("Step with id " + stepId + " not found");
}
step.failSuspendedContainer(error);
notifyStateChanged();
}
protected void setAuthorization(String key, Object value) {
this.authorizations.put(key, value);
notifyStateChanged();
@ -690,7 +715,8 @@ public class ExecutionContext implements ExecutionListener {
}
private ExecutionStatus resolveResumableStatus() {
if (!this.waitingSteps.isEmpty()) {
if (!this.waitingSteps.isEmpty()
|| this.steps.values().stream().anyMatch(step -> step.getStatus() == StepStatus.WAITING_FOR_SUBFLOW)) {
return ExecutionStatus.WAITING;
}
boolean allInputsSatisfied = this.steps.values().stream().allMatch(step ->

View File

@ -13,6 +13,7 @@ public enum ExecutionEventType {
STEP_SKIPPED,
STEP_FAILED,
STEP_WAITING_FOR_INTERACTION,
STEP_WAITING_FOR_SUBFLOW,
STEP_RESUMED,
HUMAN_DECISION_REQUESTED,
HUMAN_DECISION_RECORDED,

View File

@ -121,6 +121,7 @@ public class ExecutionObject {
this.executorService = createExecutorService(steps.size());
this.context = new ExecutionContext(steps.stream().collect(Collectors.toMap(Step::getId, Function.identity())));
this.context.getSteps().values().forEach(step -> step.setParentExecutionId(this.id));
this.context.setBiasExecutionContext(this.biasExecutionContext);
ensureRuntimeContextVariables();
registerGlobalInputs();
@ -179,6 +180,14 @@ public class ExecutionObject {
this.context.setContainerContinuation(stepId, continuation);
}
protected void completeContainerSubflow(String stepId, NodeExecutionResult result) {
this.context.completeContainerSubflow(stepId, result);
}
protected void failContainerSubflow(String stepId, String error) {
this.context.failContainerSubflow(stepId, error);
}
protected void setAuthorization(String key, Object value) {
this.providedAuthorizations.put(key, value);
this.context.setAuthorization(key, value);
@ -379,6 +388,7 @@ public class ExecutionObject {
public void restore(String executionId, long creationTime, Map<String, Object> providedAuthorizations,
ExecutionSnapshot snapshot) {
this.id = executionId;
this.context.getSteps().values().forEach(step -> step.setParentExecutionId(this.id));
if (this.runGroupId == null || this.runGroupId.isBlank()) {
this.runGroupId = executionId;
}

View File

@ -37,9 +37,11 @@ import it.cnr.isti.workflow.manager.blocks.configurations.MCPAgentChatBlockConfi
import it.cnr.isti.workflow.manager.blocks.configurations.SwitchBlockConfiguration;
import it.cnr.isti.workflow.manager.containers.Container;
import it.cnr.isti.workflow.manager.containers.configurations.ContainerConfiguration;
import it.cnr.isti.workflow.manager.containers.configurations.GenericContainerConfiguration;
import it.cnr.isti.workflow.manager.containers.configurations.LoopContainerConfiguration;
import it.cnr.isti.workflow.manager.executions.persistence.ExecutionSnapshot;
import it.cnr.isti.workflow.manager.executions.persistence.ContainerContinuationSnapshot;
import it.cnr.isti.workflow.manager.executions.steps.Step;
import it.cnr.isti.workflow.manager.executions.bias.BiasActivation;
import it.cnr.isti.workflow.manager.executions.bias.BiasApiException;
import it.cnr.isti.workflow.manager.executions.bias.BiasExecutionContext;
@ -143,9 +145,6 @@ public class ExecutionsService {
if (parentStepId == null || parentStepId.isBlank()) {
throw new IllegalArgumentException("Parent step id is required");
}
if (parentExecution.getOwner() == null || parentExecution.getOwner().isBlank()) {
throw new IllegalArgumentException("Parent execution owner is required");
}
if (parentExecution.getExecutionKind() != ExecutionKind.TOP_LEVEL) {
throw new IllegalArgumentException("Nested subflow executions are not supported");
}
@ -524,6 +523,59 @@ public class ExecutionsService {
return execution;
}
/**
* Arms the in-memory continuation used by the GenericContainer Tappa A
* runtime. The persisted continuation remains the source of identity;
* durable reconciliation after a restart is implemented in Tappa B.
*/
public void watchGenericSubflow(String parentExecutionId, String parentStepId, String childExecutionId) {
ExecutionObject child = getExecution(childExecutionId);
child.getContext().addEventListener(ignored -> reconcileGenericSubflow(
parentExecutionId, parentStepId, childExecutionId));
reconcileGenericSubflow(parentExecutionId, parentStepId, childExecutionId);
}
private synchronized void reconcileGenericSubflow(String parentExecutionId, String parentStepId,
String childExecutionId) {
ExecutionObject child = getExecution(childExecutionId);
if (!child.getContext().getStatus().isFinalState()) {
return;
}
ExecutionObject parent = getExecution(parentExecutionId);
Step<?> parentStep = parent.getContext().getSteps().get(parentStepId);
if (parentStep == null || !(parentStep.getNode() instanceof Container<?> container)
|| !(container.getSpecificConfiguration() instanceof GenericContainerConfiguration configuration)) {
return;
}
ContainerContinuationSnapshot continuation = parentStep.getContainerContinuation();
if (continuation == null || !childExecutionId.equals(continuation.getActiveInnerExecutionId())) {
return;
}
if (child.getContext().getStatus() == ExecutionStatus.SUCCESS) {
parent.setExecutionVariableDescriptors(child.getContext().getExecutionVariableDescriptors());
parent.completeContainerSubflow(parentStepId,
NodeExecutionResult.completed(resolveGenericContainerOutputs(child, configuration)));
return;
}
parent.failContainerSubflow(parentStepId,
"GenericContainer subflow ended with status " + child.getContext().getStatus()
+ (child.getContext().getErrors().isEmpty() ? "" : ": " + child.getContext().getErrors()));
}
private Map<String, Object> resolveGenericContainerOutputs(ExecutionObject child,
GenericContainerConfiguration configuration) {
Map<String, Object> outputs = new LinkedHashMap<>();
for (var exposedHandle : it.cnr.isti.workflow.manager.containers.iresolvers.ContainerFlowInterfaceResolver
.getExposedOutputs(configuration.getSubFlow())) {
var handle = exposedHandle.handle();
Object value = child.getContext().getResult().get(new FieldKey(handle.blockId(), handle.io().getName()));
if (value != null) {
outputs.put(exposedHandle.publicName(), value);
}
}
return outputs;
}
public ExecutionObject setAuthorizationValue(String executionId, String key, Object value) {
ExecutionObject eo = getExecution(executionId);
boolean knownRequirement = eo.getRequiredAuthorizations().stream()

View File

@ -7,7 +7,8 @@ import java.util.Set;
public record NodeExecutionResult(
Map<String, Object> outputs,
Set<String> notSelectedOutputs,
ExecutionOutcomeDraft outcome) {
ExecutionOutcomeDraft outcome,
Runnable onSuspended) {
public NodeExecutionResult {
outputs = outputs == null ? Map.of() : Map.copyOf(outputs);
@ -20,20 +21,33 @@ public record NodeExecutionResult(
}
public NodeExecutionResult(Map<String, Object> outputs, Set<String> notSelectedOutputs) {
this(outputs, notSelectedOutputs, null);
this(outputs, notSelectedOutputs, null, null);
}
public NodeExecutionResult(Map<String, Object> outputs, Set<String> notSelectedOutputs,
ExecutionOutcomeDraft outcome) {
this(outputs, notSelectedOutputs, outcome, null);
}
public static NodeExecutionResult completed(Map<String, Object> outputs) {
return new NodeExecutionResult(outputs, Set.of(), null);
return new NodeExecutionResult(outputs, Set.of(), null, null);
}
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, null);
return new NodeExecutionResult(outputs, notSelected, null, null);
}
public static NodeExecutionResult completedWithOutcome(Map<String, Object> outputs, ExecutionOutcomeDraft outcome) {
return new NodeExecutionResult(outputs, Set.of(), outcome);
return new NodeExecutionResult(outputs, Set.of(), outcome, null);
}
public static NodeExecutionResult suspended(Runnable onSuspended) {
return new NodeExecutionResult(Map.of(), Set.of(), null, onSuspended);
}
public boolean suspended() {
return onSuspended != null;
}
}

View File

@ -10,6 +10,7 @@ import it.cnr.isti.workflow.manager.executions.steps.Output;
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.persistence.ContainerContinuationPhase;
import it.cnr.isti.workflow.manager.flows.model.FlowNode;
import it.cnr.isti.workflow.manager.flows.model.bias.BlockBiasAnnotation;
import lombok.Builder;
@ -26,6 +27,9 @@ public class ExecutionStepView {
private StepStatus status;
private StepSkipReason skipReason;
private boolean simulated;
private String activeInnerExecutionId;
private ContainerContinuationPhase containerContinuationPhase;
private Integer containerIterationIndex;
private List<ActiveBiasProbeView> activeBiasProbes;
public static ExecutionStepView fromStep(Step<?> step) {
@ -37,6 +41,12 @@ public class ExecutionStepView {
.status(step.getStatus())
.skipReason(step.getSkipReason())
.simulated(step.isSimulated())
.activeInnerExecutionId(step.getContainerContinuation() == null
? null : step.getContainerContinuation().getActiveInnerExecutionId())
.containerContinuationPhase(step.getContainerContinuation() == null
? null : step.getContainerContinuation().getPhase())
.containerIterationIndex(step.getContainerContinuation() == null
? null : step.getContainerContinuation().getIterationIndex())
.activeBiasProbes(activeBiasProbes(step))
.build();
}

View File

@ -12,6 +12,7 @@ import it.cnr.isti.workflow.manager.blocks.configurations.HumanDecisionBlockConf
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.ContainerExecutionContext;
import it.cnr.isti.workflow.manager.executions.NodeExecutionResult;
import it.cnr.isti.workflow.manager.executions.ExecutionOutcomeDraft;
import it.cnr.isti.workflow.manager.executions.ExecutionVariableDescriptor;
@ -71,13 +72,22 @@ public final class NodeExecutors {
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, null);
}
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,
ContainerExecutionContext containerExecutionContext) {
Map<String, Object> outputs;
if (node instanceof Block<?> block) {
outputs = executeBlock(block, inputs, authorizations, executionVariables, executionVariableDescriptors,
eventLogger, biasExecutionContext);
} else if (node instanceof Container<?> container) {
outputs = executeContainer(container, inputs, authorizations, executionVariables, executionVariableDescriptors,
eventLogger, biasExecutionContext);
return executeContainer(container, inputs, authorizations, executionVariables, executionVariableDescriptors,
eventLogger, biasExecutionContext, containerExecutionContext);
} else {
throw new IllegalStateException("No executor found for node type " + node.getClass().getName());
}
@ -166,18 +176,23 @@ public final class NodeExecutors {
}
@SuppressWarnings({ "rawtypes", "unchecked" })
private static Map<String, Object> executeContainer(Container<?> container, List<Input> inputs,
private static NodeExecutionResult executeContainer(Container<?> container, List<Input> inputs,
Map<String, Object> authorizations, Map<String, Object> executionVariables,
Map<String, ExecutionVariableDescriptor> executionVariableDescriptors, ExecutionEventLogger eventLogger,
BiasExecutionContext biasExecutionContext) {
BiasExecutionContext biasExecutionContext, ContainerExecutionContext containerExecutionContext) {
BiasPreparedExecution prepared = BiasBehaviorAdapterRegistry.prepare(container, inputs, executionVariables,
biasExecutionContext, eventLogger);
Map<String, Object> result = ContainerExecutors.get(container.getType()).execute((Container) container,
NodeExecutionResult result = ContainerExecutors.get(container.getType()).execute((Container) container,
prepared.getInputs(), authorizations, prepared.getExecutionVariables(), executionVariableDescriptors,
eventLogger, biasExecutionContext);
eventLogger, biasExecutionContext, containerExecutionContext);
executionVariables.clear();
executionVariables.putAll(prepared.getExecutionVariables());
return BiasBehaviorAdapterRegistry.complete(prepared, result);
if (result.suspended()) {
return result;
}
Map<String, Object> outputs = BiasBehaviorAdapterRegistry.complete(prepared, result.outputs());
NodeExecutionResult enriched = nodeExecutionResult(container, inputs, outputs);
return new NodeExecutionResult(enriched.outputs(), enriched.notSelectedOutputs(), enriched.outcome());
}
@SuppressWarnings({ "rawtypes", "unchecked" })

View File

@ -5,16 +5,19 @@ import java.util.Map;
import it.cnr.isti.workflow.manager.containers.Container;
import it.cnr.isti.workflow.manager.executions.ExecutionEventLogger;
import it.cnr.isti.workflow.manager.executions.ContainerExecutionContext;
import it.cnr.isti.workflow.manager.containers.types.ContainerType;
import it.cnr.isti.workflow.manager.executions.ExecutionVariableDescriptor;
import it.cnr.isti.workflow.manager.executions.NodeExecutionResult;
import it.cnr.isti.workflow.manager.executions.bias.BiasExecutionContext;
import it.cnr.isti.workflow.manager.executions.steps.Input;
public interface ContainerExecutor<T extends ContainerType> {
Map<String, Object> execute(Container<T> container, List<Input> inputs, Map<String, Object> authorizations,
NodeExecutionResult execute(Container<T> container, List<Input> inputs, Map<String, Object> authorizations,
Map<String, Object> executionVariables, Map<String, ExecutionVariableDescriptor> executionVariableDescriptors,
ExecutionEventLogger eventLogger, BiasExecutionContext biasExecutionContext);
ExecutionEventLogger eventLogger, BiasExecutionContext biasExecutionContext,
ContainerExecutionContext executionContext);
Class<T> getContainerType();
}

View File

@ -13,6 +13,7 @@ import it.cnr.isti.workflow.manager.containers.configurations.GenericContainerCo
import it.cnr.isti.workflow.manager.containers.iresolvers.ContainerFlowInterfaceResolver;
import it.cnr.isti.workflow.manager.containers.types.GenericContainerType;
import it.cnr.isti.workflow.manager.executions.ExecutionEventLogger;
import it.cnr.isti.workflow.manager.executions.ContainerExecutionContext;
import it.cnr.isti.workflow.manager.executions.ExecutionEvent;
import it.cnr.isti.workflow.manager.executions.ExecutionEventType;
import it.cnr.isti.workflow.manager.executions.ExecutionObject;
@ -21,8 +22,12 @@ import it.cnr.isti.workflow.manager.executions.ExecutionStatus;
import it.cnr.isti.workflow.manager.executions.ExecutionVariableDescriptor;
import it.cnr.isti.workflow.manager.executions.ExecutionsService;
import it.cnr.isti.workflow.manager.executions.FieldKey;
import it.cnr.isti.workflow.manager.executions.ContainerSubflowRole;
import it.cnr.isti.workflow.manager.executions.NodeExecutionResult;
import it.cnr.isti.workflow.manager.executions.bias.BiasExecutionContext;
import it.cnr.isti.workflow.manager.executions.bias.runtime.BiasContainerPropagation;
import it.cnr.isti.workflow.manager.executions.persistence.ContainerContinuationPhase;
import it.cnr.isti.workflow.manager.executions.persistence.ContainerContinuationSnapshot;
import it.cnr.isti.workflow.manager.executions.steps.Input;
@Component
@ -32,10 +37,10 @@ public class GenericContainerExecutor implements ContainerExecutor<GenericContai
private ExecutionsService executionsService;
@Override
public Map<String, Object> execute(Container<GenericContainerType> container, List<Input> inputs,
public NodeExecutionResult execute(Container<GenericContainerType> container, List<Input> inputs,
Map<String, Object> authorizations, Map<String, Object> executionVariables,
Map<String, ExecutionVariableDescriptor> executionVariableDescriptors, ExecutionEventLogger eventLogger,
BiasExecutionContext biasExecutionContext) {
BiasExecutionContext biasExecutionContext, ContainerExecutionContext executionContext) {
GenericContainerConfiguration configuration = (GenericContainerConfiguration) container.getSpecificConfiguration();
if (configuration.getSubFlow() == null
|| ((configuration.getSubFlow().getBlocks() == null || configuration.getSubFlow().getBlocks().isEmpty())
@ -51,9 +56,18 @@ public class GenericContainerExecutor implements ContainerExecutor<GenericContai
BiasExecutionContext innerBiasExecutionContext = BiasContainerPropagation.innerContextFor(container.getId(),
List.of(configuration.getSubFlow()), biasExecutionContext);
ExecutionObject innerExecution = executionsService.createInnerExecution(container.getName() + " subflow",
configuration.getSubFlow(), innerBiasExecutionContext);
configuration.getSubFlow(), innerBiasExecutionContext,
executionsService.getExecution(executionContext.parentExecutionId()), executionContext.parentStepId(), null,
ContainerSubflowRole.MAIN);
forwardInnerEvents(innerExecution, eventLogger, null, GenericContainerType.TYPE);
executionsService.setContainerContinuation(executionContext.parentExecutionId(), executionContext.parentStepId(),
ContainerContinuationSnapshot.builder()
.activeInnerExecutionId(innerExecution.getId())
.phase(ContainerContinuationPhase.CHILD_CREATED)
.revision(1)
.build());
propagateAuthorizations(innerExecution, authorizations);
executionsService.setExecutionVariableDescriptors(innerExecution.getId(), executionVariableDescriptors);
executionsService.setGlobalInputDescriptors(innerExecution.getId(),
@ -81,24 +95,32 @@ public class GenericContainerExecutor implements ContainerExecutor<GenericContai
}
innerExecution = executionsService.startExecution(innerExecution.getId());
while (innerExecution.getContext().getStatus() == ExecutionStatus.RUNNING) {
try {
Thread.sleep(25L);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new IllegalStateException("Interrupted while executing GenericContainer subflow", e);
}
innerExecution = executionsService.getExecution(innerExecution.getId());
if (innerExecution.getContext().getStatus().isFinalState()) {
executionsService.setContainerContinuation(executionContext.parentExecutionId(), executionContext.parentStepId(), null);
return completedResult(innerExecution, configuration, executionVariables, executionVariableDescriptors, eventLogger);
}
executionsService.setContainerContinuation(executionContext.parentExecutionId(), executionContext.parentStepId(),
ContainerContinuationSnapshot.builder()
.activeInnerExecutionId(innerExecution.getId())
.phase(innerExecution.getContext().getStatus() == ExecutionStatus.WAITING
? ContainerContinuationPhase.WAITING_FOR_SUBFLOW
: ContainerContinuationPhase.CHILD_RUNNING)
.revision(2)
.build());
String childId = innerExecution.getId();
return NodeExecutionResult.suspended(() -> executionsService.watchGenericSubflow(
executionContext.parentExecutionId(), executionContext.parentStepId(), childId));
}
if (innerExecution.getContext().getStatus() == ExecutionStatus.WAITING) {
throw new IllegalStateException("GenericContainer subflow reached WAITING state. Interactive blocks inside GenericContainer are not supported yet");
}
private NodeExecutionResult completedResult(ExecutionObject innerExecution,
GenericContainerConfiguration configuration, Map<String, Object> executionVariables,
Map<String, ExecutionVariableDescriptor> executionVariableDescriptors, ExecutionEventLogger eventLogger) {
if (innerExecution.getContext().getStatus() == ExecutionStatus.ERROR) {
throw new IllegalStateException("GenericContainer subflow failed: " + innerExecution.getContext().getErrors());
}
if (innerExecution.getContext().getStatus() != ExecutionStatus.SUCCESS) {
throw new IllegalStateException("GenericContainer subflow ended in unexpected status: " + innerExecution.getContext().getStatus());
throw new IllegalStateException("GenericContainer subflow ended in unexpected status: "
+ innerExecution.getContext().getStatus());
}
if (eventLogger != null) {
eventLogger.info(ExecutionEventType.CONTAINER_SUBFLOW_COMPLETED,
@ -119,7 +141,7 @@ public class GenericContainerExecutor implements ContainerExecutor<GenericContai
outputs.put(exposedHandle.publicName(), value);
}
}
return outputs;
return NodeExecutionResult.completed(outputs);
}
private void propagateAuthorizations(ExecutionObject innerExecution, Map<String, Object> authorizations) {

View File

@ -17,6 +17,8 @@ import it.cnr.isti.workflow.manager.containers.configurations.IteratorContainerC
import it.cnr.isti.workflow.manager.containers.iresolvers.IteratorContainerInterfaceResolver;
import it.cnr.isti.workflow.manager.containers.types.IteratorContainerType;
import it.cnr.isti.workflow.manager.executions.ExecutionEventLogger;
import it.cnr.isti.workflow.manager.executions.ContainerExecutionContext;
import it.cnr.isti.workflow.manager.executions.NodeExecutionResult;
import it.cnr.isti.workflow.manager.executions.ExecutionEvent;
import it.cnr.isti.workflow.manager.executions.ExecutionEventType;
import it.cnr.isti.workflow.manager.executions.ExecutionObject;
@ -49,10 +51,10 @@ public class IteratorContainerExecutor implements ContainerExecutor<IteratorCont
}
@Override
public Map<String, Object> execute(Container<IteratorContainerType> container, List<Input> inputs,
public NodeExecutionResult execute(Container<IteratorContainerType> container, List<Input> inputs,
Map<String, Object> authorizations, Map<String, Object> executionVariables,
Map<String, ExecutionVariableDescriptor> executionVariableDescriptors, ExecutionEventLogger eventLogger,
BiasExecutionContext biasExecutionContext) {
BiasExecutionContext biasExecutionContext, ContainerExecutionContext executionContext) {
IteratorContainerConfiguration configuration = (IteratorContainerConfiguration) container.getSpecificConfiguration();
BiasExecutionContext innerBiasExecutionContext = BiasContainerPropagation.innerContextFor(container.getId(),
List.of(configuration.getSubFlow()), biasExecutionContext);
@ -134,7 +136,7 @@ public class IteratorContainerExecutor implements ContainerExecutor<IteratorCont
}
}
return new LinkedHashMap<>(collectedOutputs);
return NodeExecutionResult.completed(new LinkedHashMap<>(collectedOutputs));
}
private void propagateAuthorizations(ExecutionObject innerExecution, Map<String, Object> authorizations) {

View File

@ -17,6 +17,8 @@ import it.cnr.isti.workflow.manager.containers.configurations.LoopContainerConfi
import it.cnr.isti.workflow.manager.containers.iresolvers.ContainerFlowInterfaceResolver;
import it.cnr.isti.workflow.manager.containers.types.LoopContainerType;
import it.cnr.isti.workflow.manager.executions.ExecutionEventLogger;
import it.cnr.isti.workflow.manager.executions.ContainerExecutionContext;
import it.cnr.isti.workflow.manager.executions.NodeExecutionResult;
import it.cnr.isti.workflow.manager.executions.ExecutionEvent;
import it.cnr.isti.workflow.manager.executions.ExecutionEventType;
import it.cnr.isti.workflow.manager.executions.ExecutionObject;
@ -50,10 +52,10 @@ public class LoopContainerExecutor implements ContainerExecutor<LoopContainerTyp
}
@Override
public Map<String, Object> execute(Container<LoopContainerType> container, List<Input> inputs,
public NodeExecutionResult execute(Container<LoopContainerType> container, List<Input> inputs,
Map<String, Object> authorizations, Map<String, Object> executionVariables,
Map<String, ExecutionVariableDescriptor> executionVariableDescriptors, ExecutionEventLogger eventLogger,
BiasExecutionContext biasExecutionContext) {
BiasExecutionContext biasExecutionContext, ContainerExecutionContext executionContext) {
LoopContainerConfiguration configuration = (LoopContainerConfiguration) container.getSpecificConfiguration();
if (configuration.getSubFlow() == null || configuration.getSubFlow().getNodes().isEmpty()) {
throw new IllegalArgumentException("LoopContainer subFlow is empty");
@ -126,7 +128,7 @@ public class LoopContainerExecutor implements ContainerExecutor<LoopContainerTyp
if (decision.shouldContinue()) {
currentInputs.put(feedbackInput, decision.feedback());
} else {
return new LinkedHashMap<>(latestOutputs);
return NodeExecutionResult.completed(new LinkedHashMap<>(latestOutputs));
}
}

View File

@ -14,6 +14,7 @@ import it.cnr.isti.workflow.manager.blocks.Block;
import it.cnr.isti.workflow.manager.blocks.configurations.HumanDecisionBlockConfiguration;
import it.cnr.isti.workflow.manager.executions.ExecutionEventLogger;
import it.cnr.isti.workflow.manager.executions.ExecutionListener;
import it.cnr.isti.workflow.manager.executions.ContainerExecutionContext;
import it.cnr.isti.workflow.manager.executions.InteractionResult;
import it.cnr.isti.workflow.manager.executions.NodeExecutionResult;
import it.cnr.isti.workflow.manager.executions.NodeExecutionException;
@ -50,6 +51,10 @@ public class Step<N extends FlowNode> implements InputListener {
@Getter
private String id;
@Setter
@JsonIgnore
private String parentExecutionId;
@Getter
private final List<Input> inputs = new ArrayList<>();
@ -203,7 +208,14 @@ public class Step<N extends FlowNode> implements InputListener {
executionVariableDescriptors, this.interactionSimulationDescriptor, this.eventLogger,
this.biasExecutionContext)
: NodeExecutors.executeResult(this.node, this.inputs, authorizations, executionVariables,
executionVariableDescriptors, this.eventLogger, this.biasExecutionContext);
executionVariableDescriptors, this.eventLogger, this.biasExecutionContext,
new ContainerExecutionContext(this.parentExecutionId, this.id));
if (executionResult.suspended()) {
this.status = StepStatus.WAITING_FOR_SUBFLOW;
listener.paused(this.id);
executionResult.onSuspended().run();
return;
}
Map<String, Object> outputs = executionResult.outputs();
for (Output output : this.outputs) {
if (outputs.containsKey(output.getDescriptor().getName())) {
@ -329,7 +341,7 @@ public class Step<N extends FlowNode> implements InputListener {
}
private void refreshStateFromInputs() {
if (this.status == StepStatus.WAITING_FOR_INTERACTION) {
if (this.status == StepStatus.WAITING_FOR_INTERACTION || this.status == StepStatus.WAITING_FOR_SUBFLOW) {
return;
}
if (this.activationPolicy == null) {
@ -432,6 +444,34 @@ public class Step<N extends FlowNode> implements InputListener {
this.containerContinuation = null;
}
public synchronized void completeSuspendedContainer(NodeExecutionResult executionResult) {
if (this.status != StepStatus.WAITING_FOR_SUBFLOW) {
return;
}
this.status = StepStatus.RUNNING;
listener.resumed(this.id);
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.containerContinuation = null;
this.status = StepStatus.COMPLETED;
listener.completed(this.id, executionResult);
}
public synchronized void failSuspendedContainer(String error) {
if (this.status != StepStatus.WAITING_FOR_SUBFLOW) {
return;
}
this.containerContinuation = null;
this.status = StepStatus.FAILED;
listener.failed(this.id, error);
}
public ExecutionStepSnapshot snapshot() {
return ExecutionStepSnapshot.builder()
.stepId(this.id)

View File

@ -7,6 +7,7 @@ public enum StepStatus {
READY,
RUNNING,
WAITING_FOR_INTERACTION,
WAITING_FOR_SUBFLOW,
COMPLETED,
SKIPPED,
FAILED,

View File

@ -24,6 +24,7 @@ import it.cnr.isti.workflow.manager.blocks.types.HumanDecisionBlockType;
import it.cnr.isti.workflow.manager.blocks.types.SwitchBlockType;
import it.cnr.isti.workflow.manager.containers.Container;
import it.cnr.isti.workflow.manager.containers.configurations.ContainerConfiguration;
import it.cnr.isti.workflow.manager.containers.configurations.GenericContainerConfiguration;
import it.cnr.isti.workflow.manager.containers.configurations.LoopContainerConfiguration;
import it.cnr.isti.workflow.manager.containers.factories.ContainerFactory;
import it.cnr.isti.workflow.manager.containers.iresolvers.ContainerFlowInterfaceResolver;
@ -396,7 +397,8 @@ public class FlowDataValidator implements ConstraintValidator<ValidFlowStructure
throw validationError(error(ValidationErrorCode.NESTED_CONTAINERS_NOT_SUPPORTED, "container", container.getId(), fieldPath,
"Nested containers are not supported"));
}
if (nestedNodes.stream().anyMatch(FlowNode::isUserInteractive)) {
if (!(containerConfiguration instanceof GenericContainerConfiguration)
&& nestedNodes.stream().anyMatch(FlowNode::isUserInteractive)) {
throw validationError(error(ValidationErrorCode.INTERACTIVE_NODES_IN_CONTAINER_NOT_SUPPORTED, "container", container.getId(), fieldPath,
"Interactive blocks inside containers are not supported yet"));
}

View File

@ -22,6 +22,7 @@ import it.cnr.isti.workflow.manager.blocks.configurations.MCPAgentBlockConfigura
import it.cnr.isti.workflow.manager.blocks.configurations.MCPAgentChatBlockConfiguration;
import it.cnr.isti.workflow.manager.containers.Container;
import it.cnr.isti.workflow.manager.containers.configurations.ContainerConfiguration;
import it.cnr.isti.workflow.manager.containers.configurations.GenericContainerConfiguration;
import it.cnr.isti.workflow.manager.flows.model.Connection;
import it.cnr.isti.workflow.manager.flows.model.Dependency;
import it.cnr.isti.workflow.manager.flows.model.FlowData;
@ -124,15 +125,17 @@ public class FlowExecutionValidator {
"specificConfiguration.subFlow",
"Nested containers are not supported")));
subFlow.getNodes().stream()
.filter(FlowNode::isUserInteractive)
.findFirst()
.ifPresent(candidate -> errors.add(new ValidationError(
ValidationErrorCode.INTERACTIVE_NODES_IN_CONTAINER_NOT_SUPPORTED,
"container",
container.getId(),
"specificConfiguration.subFlow",
"Interactive nodes inside containers are not supported yet")));
if (!(containerConfiguration instanceof GenericContainerConfiguration)) {
subFlow.getNodes().stream()
.filter(FlowNode::isUserInteractive)
.findFirst()
.ifPresent(candidate -> errors.add(new ValidationError(
ValidationErrorCode.INTERACTIVE_NODES_IN_CONTAINER_NOT_SUPPORTED,
"container",
container.getId(),
"specificConfiguration.subFlow",
"Interactive nodes inside containers are not supported yet")));
}
}
errors.addAll(collectErrors(containerConfiguration.getSubFlow()).stream()

View File

@ -19,6 +19,7 @@ import it.cnr.isti.workflow.manager.app.ObjectMapperHolder;
import it.cnr.isti.workflow.manager.blocks.Block;
import it.cnr.isti.workflow.manager.blocks.Position;
import it.cnr.isti.workflow.manager.blocks.configurations.LLMBlockConfiguration;
import it.cnr.isti.workflow.manager.blocks.configurations.HumanInteractiveBlockConfiguration;
import it.cnr.isti.workflow.manager.blocks.configurations.SwitchBlockConfiguration;
import it.cnr.isti.workflow.manager.blocks.configurations.SwitchCase;
import it.cnr.isti.workflow.manager.blocks.types.HumanInteractionBlockType;
@ -384,16 +385,18 @@ public class ContainersControllerTest {
}
@Test
public void validateGenericContainerSubFlowRejectsInteractiveNodes() {
Block<HumanInteractionBlockType> internalBlock = blocksController.getExampleForType(HumanInteractionBlockType.TYPE);
public void validateGenericContainerSubFlowAcceptsInteractiveNodes() {
Block<HumanInteractionBlockType> internalBlock = blocksController.create(
HumanInteractiveBlockConfiguration.builder()
.name("Review")
.actionDescription("Review the submission")
.build());
ContainersController.ValidationResult result = containersController.validateContainerSubFlow(
new ContainersController.ContainerSubFlowValidationRequest(
FlowData.builder().block(internalBlock).build()));
assertFalse(result.valid());
assertTrue(result.errors().stream()
.anyMatch(error -> error.message().contains("Interactive nodes inside containers are not supported yet")));
assertTrue(result.valid());
}
@Test

View File

@ -172,7 +172,6 @@ public class SecureRetrieverControllerTest {
.findFirst()
.orElseThrow();
assertFalse(invalidItem.valid());
assertTrue(invalidItem.validationErrors().stream()
.anyMatch(error -> error.message().contains("Interactive nodes inside containers are not supported yet")));
assertFalse(invalidItem.validationErrors().isEmpty());
}
}

View File

@ -1357,6 +1357,61 @@ public class ExecutionTest {
&& event.getDetails().get("innerExecutionId") != null));
}
@Test
public void genericContainerSuspendsForInnerInteractionAndResumesFromChild() {
Block<HumanInteractionBlockType> innerInteraction = humanInteractiveBlockFactory.create(
HumanInteractiveBlockConfiguration.builder()
.name("Inner review")
.actionDescription("Review the submitted value")
.build());
Container<GenericContainerType> container = genericContainerFactory.create(GenericContainerConfiguration.builder()
.name("Interactive container")
.subFlow(FlowData.builder().block(innerInteraction).build())
.build());
ExecutionObject parent = executionsService.createExecution("Interactive container flow",
FlowData.builder().container(container).build(), "container-owner");
executionsService.prepareInput(parent.getId(), container.getId(), "input", "submission");
executionsService.startExecution(parent.getId());
parent = waitForExecutionStatus(parent.getId(), ExecutionStatus.WAITING);
Step<?> containerStep = parent.getContext().getSteps().get(container.getId());
assertEquals(StepStatus.WAITING_FOR_SUBFLOW, containerStep.getStatus());
assertNotNull(containerStep.getContainerContinuation());
String childId = containerStep.getContainerContinuation().getActiveInnerExecutionId();
ExecutionObject child = executionsService.getExecutionByOwner(childId, "container-owner");
assertEquals(ExecutionStatus.WAITING, child.getContext().getStatus());
assertEquals(StepStatus.WAITING_FOR_INTERACTION,
child.getContext().getSteps().get(innerInteraction.getId()).getStatus());
executionsService.setInteractionValue(childId, innerInteraction.getId(), "output", "approved");
parent = waitForExecutionStatus(parent.getId(), ExecutionStatus.SUCCESS);
assertEquals("approved", parent.getContext().getResult()
.get(new FieldKey(container.getId(), "output")));
assertTrue(parent.getContext().getEvents().stream()
.anyMatch(event -> event.getType() == ExecutionEventType.STEP_WAITING_FOR_SUBFLOW));
}
private ExecutionObject waitForExecutionStatus(String executionId, ExecutionStatus expectedStatus) {
long deadline = System.currentTimeMillis() + 2_000L;
ExecutionObject execution;
do {
execution = executionsService.getExecution(executionId);
if (execution.getContext().getStatus() == expectedStatus) {
return execution;
}
try {
Thread.sleep(10L);
} catch (InterruptedException exception) {
Thread.currentThread().interrupt();
throw new RuntimeException(exception);
}
} while (System.currentTimeMillis() < deadline);
assertEquals(expectedStatus, execution.getContext().getStatus());
return execution;
}
@Test
public void iteratorContainerExecutesSubflowForEachIteratedValue() {
Block<LLMBlockType> internalBlock = llmBlockFactory.create(LLMBlockConfiguration.builder()