diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionsService.java b/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionsService.java index 9b932f8..bf12487 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionsService.java +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionsService.java @@ -50,7 +50,6 @@ import it.cnr.isti.workflow.manager.executions.persistence.ExecutionSnapshot; 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.Step; -import it.cnr.isti.workflow.manager.executions.executors.BooleanLlmResponseParser; import it.cnr.isti.workflow.manager.ios.IODescriptor; import it.cnr.isti.workflow.manager.executions.bias.BiasActivation; import it.cnr.isti.workflow.manager.executions.bias.BiasApiException; @@ -915,14 +914,14 @@ public class ExecutionsService { Map inputPortsByName = ContainerFlowInterfaceResolver .getExposedInputs(configuration.getSubFlow()).stream() .collect(Collectors.toMap(ContainerFlowInterfaceResolver.ExposedHandle::publicName, handle -> handle)); - String feedbackInput = resolveFeedbackInput(configuration, inputPortsByName); + String feedbackInput = LoopGuardSupport.resolveFeedbackInput(configuration, inputPortsByName); List guardOutputHandles = ContainerFlowInterfaceResolver .getExposedOutputs(configuration.getGuardSubFlow()); Map guardResult = ContainerExecutionSupport.collectExposedOutputsAsMap(completedChild, guardOutputHandles); - boolean shouldContinue = parseLoopGuardResponse( - String.valueOf(requireGuardOutput(guardResult, LoopContainerConfiguration.GUARD_OUTPUT))); - String feedback = String.valueOf(requireGuardOutput(guardResult, LoopContainerConfiguration.FEEDBACK_OUTPUT)); - logLoopGuardEvaluation(eventLogger, iterationIndex, shouldContinue); + boolean shouldContinue = LoopGuardSupport.parseLoopGuardResponse( + String.valueOf(LoopGuardSupport.requireGuardOutput(guardResult, LoopContainerConfiguration.GUARD_OUTPUT))); + String feedback = String.valueOf(LoopGuardSupport.requireGuardOutput(guardResult, LoopContainerConfiguration.FEEDBACK_OUTPUT)); + LoopGuardSupport.logLoopGuardEvaluation(eventLogger, iterationIndex, shouldContinue); if (!shouldContinue) { setContainerContinuation(parent.getId(), parentStep.getId(), null); parent.completeContainerSubflow(parentStep.getId(), NodeExecutionResult.completed(latestOutputs)); @@ -948,7 +947,7 @@ public class ExecutionsService { Map inputPortsByName = ContainerFlowInterfaceResolver .getExposedInputs(configuration.getSubFlow()).stream() .collect(Collectors.toMap(ContainerFlowInterfaceResolver.ExposedHandle::publicName, handle -> handle)); - String feedbackInput = resolveFeedbackInput(configuration, inputPortsByName); + String feedbackInput = LoopGuardSupport.resolveFeedbackInput(configuration, inputPortsByName); List mainOutputHandles = ContainerFlowInterfaceResolver .getExposedOutputs(configuration.getSubFlow()); Map guardInputsByName = ContainerFlowInterfaceResolver @@ -990,9 +989,9 @@ public class ExecutionsService { continue; } - Map guardTemplateValues = buildGuardTemplateValues(inputs, latestOutputs, iterationIndex); + Map guardTemplateValues = LoopGuardSupport.buildGuardTemplateValues(inputs, latestOutputs, iterationIndex); Map nextInputsIfContinuing = new LinkedHashMap<>(inputs); - updateInputsForNextIteration(nextInputsIfContinuing, latestOutputs, inputPortsByName); + LoopGuardSupport.updateInputsForNextIteration(nextInputsIfContinuing, latestOutputs, inputPortsByName); ExecutionObject guardChild = createAndStartSubflowChild(parent, parentStepId, container, container.getName() + " guard iteration " + iterationIndex, configuration.getGuardSubFlow(), @@ -1008,10 +1007,10 @@ public class ExecutionsService { } parent.setExecutionVariableDescriptors(guardChild.getContext().getExecutionVariableDescriptors()); Map guardResult = ContainerExecutionSupport.collectExposedOutputsAsMap(guardChild, guardOutputHandles); - boolean shouldContinue = parseLoopGuardResponse( - String.valueOf(requireGuardOutput(guardResult, LoopContainerConfiguration.GUARD_OUTPUT))); - String feedback = String.valueOf(requireGuardOutput(guardResult, LoopContainerConfiguration.FEEDBACK_OUTPUT)); - logLoopGuardEvaluation(eventLogger, iterationIndex, shouldContinue); + boolean shouldContinue = LoopGuardSupport.parseLoopGuardResponse( + String.valueOf(LoopGuardSupport.requireGuardOutput(guardResult, LoopContainerConfiguration.GUARD_OUTPUT))); + String feedback = String.valueOf(LoopGuardSupport.requireGuardOutput(guardResult, LoopContainerConfiguration.FEEDBACK_OUTPUT)); + LoopGuardSupport.logLoopGuardEvaluation(eventLogger, iterationIndex, shouldContinue); if (!shouldContinue) { setContainerContinuation(parent.getId(), parentStepId, null); return ContainerAdvanceOutcome.completed(new LinkedHashMap<>(latestOutputs)); @@ -1024,78 +1023,6 @@ public class ExecutionsService { } } - private void logLoopGuardEvaluation(ExecutionEventLogger eventLogger, int iterationIndex, boolean shouldContinue) { - if (eventLogger == null) { - return; - } - eventLogger.info(ExecutionEventType.CONTAINER_CONDITION_EVALUATED, "Evaluated loop guard", - Map.of("containerType", LoopContainerType.TYPE, "iteration", iterationIndex, "continue", shouldContinue)); - eventLogger.info(ExecutionEventType.CONTAINER_ITERATION_COMPLETED, - shouldContinue ? "Completed loop iteration " + iterationIndex : "Loop completed at iteration " + iterationIndex, - Map.of("containerType", LoopContainerType.TYPE, "iteration", iterationIndex, "continue", shouldContinue)); - } - - private static String resolveFeedbackInput(LoopContainerConfiguration configuration, - Map inputPortsByName) { - if (configuration.getFeedbackInput() != null && !configuration.getFeedbackInput().isBlank()) { - String configuredInput = configuration.getFeedbackInput(); - ContainerFlowInterfaceResolver.ExposedHandle handle = inputPortsByName.get(configuredInput); - if (handle == null || handle.handle().io().isMultiple()) { - throw new IllegalArgumentException( - "LoopContainer feedbackInput must target an open non-multiple subFlow input: " + configuredInput); - } - return configuredInput; - } - List eligibleInputs = inputPortsByName.values().stream() - .filter(handle -> !handle.handle().io().isMultiple()) - .toList(); - if (eligibleInputs.size() == 1) { - return eligibleInputs.getFirst().publicName(); - } - if (eligibleInputs.isEmpty()) { - throw new IllegalArgumentException( - "LoopContainer subFlow must expose at least one open non-multiple input to receive guard feedback"); - } - throw new IllegalArgumentException( - "LoopContainer feedbackInput is required when subFlow exposes more than one open non-multiple input"); - } - - private static void updateInputsForNextIteration(Map currentInputs, Map latestOutputs, - Map inputPortsByName) { - for (Map.Entry output : latestOutputs.entrySet()) { - if (inputPortsByName.containsKey(output.getKey())) { - currentInputs.put(output.getKey(), output.getValue()); - } - } - } - - private static Map buildGuardTemplateValues(Map currentInputs, - Map latestOutputs, int iteration) { - Map values = new LinkedHashMap<>(); - if (currentInputs != null) { - values.putAll(currentInputs); - currentInputs.forEach((key, value) -> values.put("inputs." + key, value)); - } - if (latestOutputs != null) { - values.putAll(latestOutputs); - latestOutputs.forEach((key, value) -> values.put("outputs." + key, value)); - } - values.put("iteration", iteration); - return values; - } - - private static Object requireGuardOutput(Map guardResult, String outputName) { - Object value = guardResult.get(outputName); - if (value == null) { - throw new IllegalArgumentException("LoopContainer guardSubFlow must return output: " + outputName); - } - return value; - } - - private static boolean parseLoopGuardResponse(String response) { - return BooleanLlmResponseParser.parse(response, "loop guard subflow", "Loop"); - } - // ---- Shared container-child helpers ---- private NodeExecutionResult toNodeExecutionResult(ContainerAdvanceOutcome outcome, String parentExecutionId, diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/LoopGuardSupport.java b/src/main/java/it/cnr/isti/workflow/manager/executions/LoopGuardSupport.java new file mode 100644 index 0000000..2475592 --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/LoopGuardSupport.java @@ -0,0 +1,88 @@ +package it.cnr.isti.workflow.manager.executions; + +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import it.cnr.isti.workflow.manager.containers.configurations.LoopContainerConfiguration; +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.executors.BooleanLlmResponseParser; + +final class LoopGuardSupport { + + private LoopGuardSupport() { + } + + static void logLoopGuardEvaluation(ExecutionEventLogger eventLogger, int iterationIndex, boolean shouldContinue) { + if (eventLogger == null) { + return; + } + eventLogger.info(ExecutionEventType.CONTAINER_CONDITION_EVALUATED, "Evaluated loop guard", + Map.of("containerType", LoopContainerType.TYPE, "iteration", iterationIndex, "continue", shouldContinue)); + eventLogger.info(ExecutionEventType.CONTAINER_ITERATION_COMPLETED, + shouldContinue ? "Completed loop iteration " + iterationIndex : "Loop completed at iteration " + iterationIndex, + Map.of("containerType", LoopContainerType.TYPE, "iteration", iterationIndex, "continue", shouldContinue)); + } + + static String resolveFeedbackInput(LoopContainerConfiguration configuration, + Map inputPortsByName) { + if (configuration.getFeedbackInput() != null && !configuration.getFeedbackInput().isBlank()) { + String configuredInput = configuration.getFeedbackInput(); + ContainerFlowInterfaceResolver.ExposedHandle handle = inputPortsByName.get(configuredInput); + if (handle == null || handle.handle().io().isMultiple()) { + throw new IllegalArgumentException( + "LoopContainer feedbackInput must target an open non-multiple subFlow input: " + configuredInput); + } + return configuredInput; + } + List eligibleInputs = inputPortsByName.values().stream() + .filter(handle -> !handle.handle().io().isMultiple()) + .toList(); + if (eligibleInputs.size() == 1) { + return eligibleInputs.getFirst().publicName(); + } + if (eligibleInputs.isEmpty()) { + throw new IllegalArgumentException( + "LoopContainer subFlow must expose at least one open non-multiple input to receive guard feedback"); + } + throw new IllegalArgumentException( + "LoopContainer feedbackInput is required when subFlow exposes more than one open non-multiple input"); + } + + static void updateInputsForNextIteration(Map currentInputs, Map latestOutputs, + Map inputPortsByName) { + for (Map.Entry output : latestOutputs.entrySet()) { + if (inputPortsByName.containsKey(output.getKey())) { + currentInputs.put(output.getKey(), output.getValue()); + } + } + } + + static Map buildGuardTemplateValues(Map currentInputs, + Map latestOutputs, int iteration) { + Map values = new LinkedHashMap<>(); + if (currentInputs != null) { + values.putAll(currentInputs); + currentInputs.forEach((key, value) -> values.put("inputs." + key, value)); + } + if (latestOutputs != null) { + values.putAll(latestOutputs); + latestOutputs.forEach((key, value) -> values.put("outputs." + key, value)); + } + values.put("iteration", iteration); + return values; + } + + static Object requireGuardOutput(Map guardResult, String outputName) { + Object value = guardResult.get(outputName); + if (value == null) { + throw new IllegalArgumentException("LoopContainer guardSubFlow must return output: " + outputName); + } + return value; + } + + static boolean parseLoopGuardResponse(String response) { + return BooleanLlmResponseParser.parse(response, "loop guard subflow", "Loop"); + } +}