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 c2ba824..217f1d6 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 @@ -17,12 +17,15 @@ import java.util.logging.Logger; import com.fasterxml.jackson.annotation.JsonIgnore; import it.cnr.isti.workflow.manager.executions.persistence.ExecutionSnapshot; +import it.cnr.isti.workflow.manager.executions.persistence.ExecutionStepSnapshot; import it.cnr.isti.workflow.manager.executions.persistence.ContainerContinuationSnapshot; 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.flows.loops.FlowLoops; +import it.cnr.isti.workflow.manager.flows.validation.ValidationErrorCode; import it.cnr.isti.workflow.manager.llms.LLMDescriptor; import lombok.AccessLevel; import lombok.Getter; @@ -87,6 +90,18 @@ public class ExecutionContext implements ExecutionListener { List waitingSteps = new ArrayList<>(); + /** The loops this execution may go round, by the id of the step that guards each. */ + @JsonIgnore + @Getter(AccessLevel.NONE) + private final Map loopsByGuard = new HashMap<>(); + + /** + * Each loop step as it was when its round ended, oldest first. The step itself only ever holds + * the round in progress, so this is where earlier rounds - and a person's earlier verdicts - + * remain to be read. + */ + List stepHistory = new CopyOnWriteArrayList<>(); + @JsonIgnore Runnable stateChangeListener; @@ -224,6 +239,14 @@ public class ExecutionContext implements ExecutionListener { "Completed step " + resolveNodeName(id), Map.of()); clearPartialResults(id); Step completedStep = this.steps.get(id); + LoopRuntime loop = this.loopsByGuard.get(id); + if (loop != null && result.containsKey(loop.backOutput())) { + // Not a finished step with outputs to settle: the guard's other outputs are still + // open, since a later round may leave through them. + goRound(loop, result.get(loop.backOutput())); + notifyStateChanged(); + return; + } completedStep.getOutputs().forEach(output -> { String outputName = output.getDescriptor().getName(); if (result.containsKey(outputName)) { @@ -250,6 +273,59 @@ public class ExecutionContext implements ExecutionListener { notifyStateChanged(); } + /** + * Starts the next round of a loop whose guard has just chosen to go round. + * + *

Every step of the loop has finished by now: each one leads to the guard, so the guard + * cannot have run before them. That is what makes resetting them safe - nothing of the loop is + * still running to be caught half-way. The reset and the delivery of the new value happen + * under one lock, so no snapshot sees a loop that is neither in its old round nor its new one. + */ + private synchronized void goRound(LoopRuntime runtime, Object value) { + FlowLoops.Loop loop = runtime.loop(); + Step guard = this.steps.get(loop.guardId()); + int finished = guard.getIteration(); + if (finished >= loop.maxIterations()) { + String message = "Loop guarded by " + resolveNodeName(loop.guardId()) + " chose to go round again after " + + finished + " iteration" + (finished == 1 ? "" : "s") + ", which is its limit (" + + loop.maxIterations() + "). Raise the limit on the connection that leads back, " + + "or check why the loop does not converge"; + guard.failAfterCompleting(); + addStepEvent(loop.guardId(), ExecutionEventLevel.ERROR, ExecutionEventType.LOOP_LIMIT_REACHED, message, + Map.of("iteration", finished, "maxIterations", loop.maxIterations())); + failed(loop.guardId(), message, ValidationErrorCode.LOOP_ITERATION_LIMIT_REACHED.name()); + return; + } + List> members = loop.memberIds().stream().>map(this.steps::get).toList(); + members.forEach(member -> this.stepHistory.add(member.snapshot())); + for (Step member : members) { + member.rearmForNextIteration( + input -> runtime.refilledInputs().contains(new FieldKey(member.getId(), input.getDescriptor().getName())), + loop.memberIds()); + clearPartialResults(member.getId()); + } + addStepEvent(loop.entryId(), ExecutionEventLevel.INFO, ExecutionEventType.LOOP_ITERATION_STARTED, + "Going round again from " + resolveNodeName(loop.entryId()) + ": iteration " + (finished + 1) + + " of at most " + loop.maxIterations(), + Map.of("iteration", finished + 1, "maxIterations", loop.maxIterations(), "guardId", loop.guardId())); + String entryInput = loop.backEdge().getTargetName(); + this.steps.get(loop.entryId()).getInputs().stream() + .filter(input -> input.getDescriptor().getName().equals(entryInput)) + .findFirst() + .orElseThrow(() -> new IllegalStateException("Loop entry " + loop.entryId() + " has no input " + entryInput)) + .setValue(value); + } + + /** Tells this execution which loops it may go round; done once, when its steps are built. */ + public void configureLoops(List loops) { + this.loopsByGuard.clear(); + loops.forEach(loop -> this.loopsByGuard.put(loop.loop().guardId(), loop)); + } + + private boolean isInLoop(String stepId) { + return this.loopsByGuard.values().stream().anyMatch(loop -> loop.loop().contains(stepId)); + } + @Override public void skipped(String id) { skipped(id, StepSkipReason.INPUT_UNAVAILABLE); @@ -632,6 +708,7 @@ public class ExecutionContext implements ExecutionListener { .events(List.copyOf(this.events)) .waitingSteps(List.copyOf(this.waitingSteps)) .steps(this.steps.values().stream().map(Step::snapshot).toList()) + .stepHistory(List.copyOf(this.stepHistory)) .build(); } @@ -718,6 +795,10 @@ public class ExecutionContext implements ExecutionListener { if (snapshot.getWaitingSteps() != null) { this.waitingSteps.addAll(snapshot.getWaitingSteps()); } + this.stepHistory.clear(); + if (snapshot.getStepHistory() != null) { + this.stepHistory.addAll(snapshot.getStepHistory()); + } this.projectContext.clear(); if (snapshot.getProjectContext() != null) { this.projectContext.putAll(snapshot.getProjectContext()); @@ -842,6 +923,11 @@ public class ExecutionContext implements ExecutionListener { private void addStepEvent(String stepId, ExecutionEventLevel level, ExecutionEventType type, String message, Map details) { Step step = this.steps.get(stepId); + Map eventDetails = new HashMap<>(details == null ? Map.of() : details); + // A step in a loop runs once per round; without the round, its events cannot be told apart. + if (step != null && isInLoop(stepId)) { + eventDetails.putIfAbsent("iteration", step.getIteration()); + } addEvent(ExecutionEvent.builder() .stepId(stepId) .nodeId(stepId) @@ -849,7 +935,7 @@ public class ExecutionContext implements ExecutionListener { .level(level) .type(type) .message(message) - .details(details == null ? Map.of() : Map.copyOf(details)) + .details(Map.copyOf(eventDetails)) .build()); } diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionEventType.java b/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionEventType.java index 3918d3f..13fc084 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionEventType.java +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionEventType.java @@ -29,6 +29,10 @@ public enum ExecutionEventType { HUMAN_EVALUATION_REFERENCE_REVEALED, FLOW_PATH_ENDED, FLOW_OUTCOME_RECORDED, + /** A loop's guard chose to go round again, and the loop's steps were reset for the next round. */ + LOOP_ITERATION_STARTED, + /** A loop's guard chose to go round again with no rounds left; the execution fails. */ + LOOP_LIMIT_REACHED, CONTAINER_SUBFLOW_STARTED, CONTAINER_SUBFLOW_COMPLETED, CONTAINER_ITERATION_STARTED, 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 a2ec37d..a7aa2c3 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 @@ -29,6 +29,7 @@ import it.cnr.isti.workflow.manager.executions.steps.Input; import it.cnr.isti.workflow.manager.executions.steps.Output; import it.cnr.isti.workflow.manager.executions.steps.Step; import it.cnr.isti.workflow.manager.flows.model.FlowNode; +import it.cnr.isti.workflow.manager.flows.loops.FlowLoops; 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; @@ -134,11 +135,15 @@ public class ExecutionObject { this.projectRunId = projectRunId; this.projectRunOrder = projectRunOrder; - List> steps = getStepsFromFlow(flow); + FlowLoops.Analysis loops = FlowLoops.analyze(flow); + List> steps = getStepsFromFlow(flow, loops); this.executorService = createExecutorService(steps.size()); this.context = new ExecutionContext(steps.stream().collect(Collectors.toMap(Step::getId, Function.identity()))); + this.context.configureLoops(loops.loops().stream() + .map(loop -> LoopRuntime.of(loop, this.stepConnections)) + .toList()); this.context.getSteps().values().forEach(step -> step.setParentExecutionId(this.id)); this.context.setBiasExecutionContext(this.biasExecutionContext); this.context.setProjectContext(projectContext); @@ -154,7 +159,7 @@ public class ExecutionObject { } - List> getStepsFromFlow(FlowData flow) { + List> getStepsFromFlow(FlowData flow, FlowLoops.Analysis loops) { List> steps = new ArrayList<>(); flow.getNodes().stream().filter(node -> !NodeExecutors.hasExecutor(node)).findAny().ifPresent(node -> { throw new IllegalStateException("No executor found for node " + node.getName() + ". Cannot create execution"); @@ -164,6 +169,14 @@ public class ExecutionObject { for (Connection connection: connections){ Step sourceStep = steps.stream().filter(s -> s.getId().equals(connection.getSourceId())).findFirst().orElseThrow(); Step targetStep = steps.stream().filter(s -> s.getId().equals(connection.getTargetId())).findFirst().orElseThrow(); + if (loops.loops().stream().anyMatch(loop -> loop.backEdge() == connection)) { + // Not wired like the others. A wired output pushes its value the moment it is + // produced, and the loop must first be reset for the round that value starts; the + // context delivers it once it has. Left unwired, the entry's input also still + // counts as open when nothing else feeds it, so the first round asks for it. + sourceStep.setLoopBackOutput(connection.getSourceName()); + continue; + } Output output = sourceStep.getOutputs().stream().filter(o -> o.getDescriptor().getName().equals(connection.getSourceName())).findFirst().orElseThrow(); Input input = targetStep.getInputs().stream().filter(i -> i.getDescriptor().getName().equals(connection.getTargetName())).findFirst().orElseThrow(); output.registerSource(input); diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/LoopRuntime.java b/src/main/java/it/cnr/isti/workflow/manager/executions/LoopRuntime.java new file mode 100644 index 0000000..39e9647 --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/LoopRuntime.java @@ -0,0 +1,38 @@ +// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii - ISTI-CNR +// SPDX-License-Identifier: AGPL-3.0-or-later +// Attribution term under AGPL-3.0 section 7(b): see LICENSE-ADDENDUM. + +package it.cnr.isti.workflow.manager.executions; + +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +import it.cnr.isti.workflow.manager.flows.loops.FlowLoops; +import it.cnr.isti.workflow.manager.flows.model.Connection; + +/** + * A loop as an execution runs it: the loop itself, and which inputs each new round fills again. + * + * @param loop the loop, as the flow's shape defines it + * @param refilledInputs the inputs of the loop's steps that the next round delivers again - those + * fed from another step of the loop, and the one the back edge leads to. + * Every other input keeps its value from round to round. + */ +public record LoopRuntime(FlowLoops.Loop loop, Set refilledInputs) { + + public static LoopRuntime of(FlowLoops.Loop loop, List connections) { + Set refilled = new HashSet<>(); + for (Connection connection : connections) { + if (connection == loop.backEdge() + || (loop.contains(connection.getSourceId()) && loop.contains(connection.getTargetId()))) { + refilled.add(new FieldKey(connection.getTargetId(), connection.getTargetName())); + } + } + return new LoopRuntime(loop, Set.copyOf(refilled)); + } + + public String backOutput() { + return loop.backEdge().getSourceName(); + } +} 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 1ff309f..b272d1e 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 @@ -50,5 +50,7 @@ public class ExecutionSnapshot { private List events; private List waitingSteps; private List steps; + /** Loop steps as each round left them; null in snapshots written before loops existed. */ + private List stepHistory; private List outcomes; } 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 1a77f55..9cd2205 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 @@ -28,4 +28,6 @@ public class ExecutionStepSnapshot { private Map partialResults; private StepSkipReason skipReason; private ContainerContinuationSnapshot containerContinuation; + /** Which round of its loop the step was on; null in snapshots written before loops existed. */ + private Integer iteration; } 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 5d0f464..1e33e38 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 @@ -76,6 +76,16 @@ public class Input { this.registered = true; } + /** + * Forgets this input's value so the next round of a loop can deliver a new one. Only a loop + * going round does this, and only for inputs fed from inside the loop: an input's state + * otherwise never goes back to pending, and nothing else may rely on it doing so. + */ + void resetForNextIteration() { + this.value = null; + this.resolutionState = InputResolutionState.PENDING; + } + public boolean isSet() { return this.resolutionState == InputResolutionState.VALUE; } 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 2bb4386..f924684 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 @@ -10,6 +10,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.function.Predicate; import java.util.concurrent.ExecutorService; import com.fasterxml.jackson.annotation.JsonIgnore; @@ -149,6 +150,25 @@ public class Step implements InputListener { @JsonIgnore private StepActivationPolicy activationPolicy; + /** + * Which round of its loop this step is on; 1 for a step that is in no loop, which only ever + * has the one. + */ + @Getter + private int iteration = 1; + + /** + * For a loop's guard, the output that goes round. Choosing it decides nothing about the + * guard's other outputs - a later round may still leave through them - so they are not marked + * as branches not taken. + */ + @JsonIgnore + @Setter + private String loopBackOutput; + + @JsonIgnore + private final Set skippedDependencyIds = new LinkedHashSet<>(); + @Builder public Step(@NonNull N node) { // Initialize the step with the provided block @@ -235,10 +255,12 @@ public class Step implements InputListener { return; } Map outputs = executionResult.outputs(); + boolean goingRound = goesRound(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())) { + } else if (!goingRound + && executionResult.notSelectedOutputs().contains(output.getDescriptor().getName())) { output.markNotSelected(); } } @@ -287,11 +309,14 @@ public class Step implements InputListener { return; } Map resolvedOutputs = interactionResult.outputs(); + boolean goingRound = goesRound(resolvedOutputs); for (Output output : this.outputs) { if (resolvedOutputs.containsKey(output.getDescriptor().getName())) { output.setValue(resolvedOutputs.get(output.getDescriptor().getName())); } else if (interactionResult.notSelectedOutputs().contains(output.getDescriptor().getName())) { - output.markNotSelected(); + if (!goingRound) { + output.markNotSelected(); + } } else { throw new IllegalArgumentException("Missing output value for: " + output.getDescriptor().getName()); } @@ -329,6 +354,7 @@ public class Step implements InputListener { public synchronized void dependencySkipped(String stepId, StepSkipReason upstreamSkipReason) { if (this.dependencyIds.contains(stepId)) { + this.skippedDependencyIds.add(stepId); this.dependencySkipReason = upstreamSkipReason == StepSkipReason.BRANCH_NOT_SELECTED ? StepSkipReason.BRANCH_NOT_SELECTED : StepSkipReason.DEPENDENCY_UNAVAILABLE; @@ -338,6 +364,7 @@ public class Step implements InputListener { public synchronized void resetDependencyState() { this.satisfiedDependencyIds.clear(); + this.skippedDependencyIds.clear(); this.dependencySkipReason = null; refreshStateFromInputs(); } @@ -350,6 +377,41 @@ public class Step implements InputListener { return List.copyOf(this.dependentSteps); } + private boolean goesRound(Map producedOutputs) { + return this.loopBackOutput != null && producedOutputs.containsKey(this.loopBackOutput); + } + + /** + * Makes this step ready to run again, for the next round of the loop it is in. + * + *

Only the inputs {@code resetInput} selects are forgotten: those fed from inside the loop, + * which the next round will fill again. Inputs from outside keep their values, because nothing + * will send them twice. Dependencies on other steps of the loop are forgotten for the same + * reason; dependencies on steps outside it stand. + * + * @param resetInput which inputs the next round will fill + * @param loopMemberIds the steps that run again with this one + */ + public synchronized void rearmForNextIteration(Predicate resetInput, Set loopMemberIds) { + this.iteration++; + this.status = StepStatus.WAITING_FOR_INPUT; + this.skipReason = null; + this.partialResults.clear(); + this.containerContinuation = null; + this.inputs.stream().filter(resetInput).forEach(Input::resetForNextIteration); + this.satisfiedDependencyIds.removeAll(loopMemberIds); + this.skippedDependencyIds.removeAll(loopMemberIds); + if (this.skippedDependencyIds.isEmpty()) { + this.dependencySkipReason = null; + } + refreshStateFromInputs(); + } + + /** A guard that finished its round but may not start another: the loop ran out of rounds. */ + public synchronized void failAfterCompleting() { + this.status = StepStatus.FAILED; + } + private void refreshStateFromInputs() { if (this.status == StepStatus.WAITING_FOR_INTERACTION || this.status == StepStatus.WAITING_FOR_SUBFLOW) { return; @@ -497,6 +559,7 @@ public class Step implements InputListener { .partialResults(Map.copyOf(this.partialResults)) .skipReason(this.skipReason) .containerContinuation(this.containerContinuation == null ? null : this.containerContinuation.copy()) + .iteration(this.iteration) .build(); } @@ -518,6 +581,7 @@ public class Step implements InputListener { this.started = snapshot.isStarted(); this.simulated = snapshot.isSimulated(); this.skipReason = snapshot.getSkipReason(); + this.iteration = snapshot.getIteration() == null || snapshot.getIteration() < 1 ? 1 : snapshot.getIteration(); this.containerContinuation = snapshot.getContainerContinuation() == null ? null : snapshot.getContainerContinuation().copy(); diff --git a/src/main/java/it/cnr/isti/workflow/manager/flows/loops/FlowLoops.java b/src/main/java/it/cnr/isti/workflow/manager/flows/loops/FlowLoops.java index a226635..8538c7e 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/flows/loops/FlowLoops.java +++ b/src/main/java/it/cnr/isti/workflow/manager/flows/loops/FlowLoops.java @@ -210,13 +210,21 @@ public final class FlowLoops { } /** - * Of the connections that could close the loop, the one that leads to where rounds start: the - * block the loop is entered at from outside. With a single candidate there is nothing to choose. + * Of the connections that could close the loop, the one it goes round by. With a single + * candidate there is nothing to choose. Otherwise the author's word decides - the connection + * they set a limit on is the one they drew going back - and failing that, the one leading to + * where the loop is entered from outside, since that is where each round starts. */ private static Edge pickBackEdge(List closing, Set component, List edges) { if (closing.size() == 1) { return closing.getFirst(); } + List markedByTheAuthor = closing.stream() + .filter(edge -> edge.connection().getLoop() != null) + .toList(); + if (markedByTheAuthor.size() == 1) { + return markedByTheAuthor.getFirst(); + } Set enteredFromOutside = edges.stream() .filter(edge -> !component.contains(edge.sourceId()) && component.contains(edge.targetId())) .map(Edge::targetId) diff --git a/src/test/java/it/cnr/isti/workflow/manager/executions/LoopExecutionTest.java b/src/test/java/it/cnr/isti/workflow/manager/executions/LoopExecutionTest.java new file mode 100644 index 0000000..b9db23d --- /dev/null +++ b/src/test/java/it/cnr/isti/workflow/manager/executions/LoopExecutionTest.java @@ -0,0 +1,331 @@ +// SPDX-FileCopyrightText: 2025-2026 Lucio Lelii - ISTI-CNR +// SPDX-License-Identifier: AGPL-3.0-or-later +// Attribution term under AGPL-3.0 section 7(b): see LICENSE-ADDENDUM. + +package it.cnr.isti.workflow.manager.executions; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.time.Duration; +import java.time.Instant; +import java.util.List; + +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.context.TestConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.test.context.TestPropertySource; + +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.EndBlockConfiguration; +import it.cnr.isti.workflow.manager.blocks.configurations.HumanDecisionBlockConfiguration; +import it.cnr.isti.workflow.manager.blocks.configurations.HumanDecisionOption; +import it.cnr.isti.workflow.manager.blocks.configurations.LLMBlockConfiguration; +import it.cnr.isti.workflow.manager.blocks.factories.ConditionalBlockFactory; +import it.cnr.isti.workflow.manager.blocks.factories.EndBlockFactory; +import it.cnr.isti.workflow.manager.blocks.factories.HumanDecisionBlockFactory; +import it.cnr.isti.workflow.manager.blocks.factories.LLMBlockFactory; +import it.cnr.isti.workflow.manager.blocks.types.ConditionalBlockType; +import it.cnr.isti.workflow.manager.blocks.types.EndBlockType; +import it.cnr.isti.workflow.manager.blocks.types.HumanDecisionBlockType; +import it.cnr.isti.workflow.manager.blocks.types.LLMBlockType; +import it.cnr.isti.workflow.manager.executions.persistence.ExecutionStepSnapshot; +import it.cnr.isti.workflow.manager.executions.steps.StepStatus; +import it.cnr.isti.workflow.manager.flows.model.Connection; +import it.cnr.isti.workflow.manager.flows.model.FlowData; +import it.cnr.isti.workflow.manager.flows.model.LoopEdgeSettings; +import it.cnr.isti.workflow.manager.flows.validation.ValidationErrorCode; +import it.cnr.isti.workflow.manager.llms.ChatMessage; +import it.cnr.isti.workflow.manager.llms.LLMDescriptor; +import it.cnr.isti.workflow.manager.llms.providers.LLMProvider; + +/** + * Running a flow that goes round: a draft is written, checked, and sent back until it passes. + * + *

The model here only repeats what it is given between {@code <<} and {@code >>}. The draft's + * prompt adds a suffix to the topic, so each round's answer is one suffix longer than the last, + * and the check passes once it is long enough - a loop whose number of rounds is known in advance. + */ +@SpringBootTest +@TestPropertySource(locations = "classpath:test.properties") +class LoopExecutionTest { + + private static final LLMDescriptor ECHO = LLMDescriptor.builder().provider("loopEcho").model("echo").build(); + + @TestConfiguration + static class EchoModel { + + @Bean + LLMProvider loopEchoProvider() { + return new LLMProvider() { + + @Override + public String getName() { + return "loopEcho"; + } + + @Override + public List getRegisteredModels() { + return List.of("echo"); + } + + @Override + public String generate(String model, String prompt) { + int start = prompt.indexOf("<<"); + int end = prompt.indexOf(">>", start); + return start < 0 || end < 0 ? prompt : prompt.substring(start + 2, end); + } + + @Override + public String chat(String model, List messages) { + return generate(model, messages.getLast().content()); + } + }; + } + } + + @Autowired + ExecutionsService executionsService; + + @Autowired + LLMBlockFactory llmBlockFactory; + + @Autowired + ConditionalBlockFactory conditionalBlockFactory; + + @Autowired + HumanDecisionBlockFactory humanDecisionBlockFactory; + + @Autowired + EndBlockFactory endBlockFactory; + + @Test + void aDraftGoesRoundUntilTheCheckPassesAndThenLeaves() { + RevisionLoop loop = revisionLoop("${{response}}.length() >= 4", null); + + ExecutionObject execution = start(loop, "a"); + execution = awaitFinal(execution.getId()); + + assertEquals(ExecutionStatus.SUCCESS, execution.getContext().getStatus(), () -> errors(loop, executionFor(loop))); + // "a" -> "ax" (too short) -> "axx" (too short) -> "axxx": three rounds. + assertEquals("axxx", execution.getContext().getOutcomes().getFirst().payload()); + assertEquals(3, execution.getContext().getSteps().get(loop.draft().getId()).getIteration()); + assertEquals(3, execution.getContext().getSteps().get(loop.check().getId()).getIteration()); + assertEquals(1, execution.getContext().getSteps().get(loop.accepted().getId()).getIteration(), + "the step after the loop runs once, after it"); + assertEquals(StepStatus.COMPLETED, execution.getContext().getSteps().get(loop.accepted().getId()).getStatus()); + } + + @Test + void anInputFromOutsideTheLoopKeepsItsValueOnEveryRound() { + // The suffix is given once, at the start. If a new round forgot it, the second draft would + // wait for a value nobody is going to send again. + RevisionLoop loop = revisionLoop("${{response}}.length() >= 7", null); + + ExecutionObject execution = executionsService.createExecution("Loop keeps invariants", loop.flow()); + executionsService.prepareInput(execution.getId(), loop.draft().getId(), "topic", "a"); + executionsService.prepareInput(execution.getId(), loop.draft().getId(), "suffix", "xy"); + executionsService.startExecution(execution.getId()); + execution = awaitFinal(execution.getId()); + + assertEquals(ExecutionStatus.SUCCESS, execution.getContext().getStatus()); + assertEquals("axyxyxy", execution.getContext().getOutcomes().getFirst().payload()); + } + + @Test + void earlierRoundsStayReadableAfterTheLoopMovesOn() { + RevisionLoop loop = revisionLoop("${{response}}.length() >= 4", null); + + ExecutionObject execution = awaitFinal(start(loop, "a").getId()); + + List draftRounds = execution.getContext().getStepHistory().stream() + .filter(snapshot -> snapshot.getStepId().equals(loop.draft().getId())) + .toList(); + assertEquals(List.of(1, 2), draftRounds.stream().map(ExecutionStepSnapshot::getIteration).toList(), + "two rounds ended by going round; the third is the step itself"); + assertEquals(List.of("a", "ax"), draftRounds.stream() + .map(snapshot -> snapshot.getInputs().stream() + .filter(input -> input.getName().equals("topic")) + .findFirst().orElseThrow().getValue()) + .toList()); + + List rounds = execution.getContext().getEvents().stream() + .filter(event -> event.getType() == ExecutionEventType.LOOP_ITERATION_STARTED) + .toList(); + assertEquals(List.of(2, 3), rounds.stream().map(event -> event.getDetails().get("iteration")).toList()); + assertEquals(List.of(1, 2, 3), execution.getContext().getEvents().stream() + .filter(event -> event.getType() == ExecutionEventType.STEP_COMPLETED) + .filter(event -> loop.draft().getId().equals(event.getStepId())) + .map(event -> event.getDetails().get("iteration")) + .toList(), "each round's events say which round they belong to"); + } + + @Test + void aLoopThatNeverPassesFailsAtItsLimitInsteadOfLeaving() { + RevisionLoop loop = revisionLoop("${{response}}.length() > 1000", new LoopEdgeSettings(3)); + + ExecutionObject execution = awaitFinal(start(loop, "a").getId()); + + assertEquals(ExecutionStatus.ERROR, execution.getContext().getStatus()); + assertEquals(ValidationErrorCode.LOOP_ITERATION_LIMIT_REACHED.name(), + execution.getContext().getErrorCodes().get(loop.check().getId())); + assertTrue(execution.getContext().getErrors().get(loop.check().getId()).contains("3"), + execution.getContext().getErrors().get(loop.check().getId())); + assertEquals(3, execution.getContext().getSteps().get(loop.check().getId()).getIteration()); + assertEquals(StepStatus.FAILED, execution.getContext().getSteps().get(loop.check().getId()).getStatus()); + assertNotEquals(StepStatus.COMPLETED, execution.getContext().getSteps().get(loop.accepted().getId()).getStatus(), + "reaching the limit is not a way out"); + assertTrue(execution.getContext().getEvents().stream() + .anyMatch(event -> event.getType() == ExecutionEventType.LOOP_LIMIT_REACHED)); + } + + @Test + void withoutASettingTheLimitIsTen() { + RevisionLoop loop = revisionLoop("${{response}}.length() > 1000", null); + + ExecutionObject execution = awaitFinal(start(loop, "a").getId()); + + assertEquals(ExecutionStatus.ERROR, execution.getContext().getStatus()); + assertEquals(10, execution.getContext().getSteps().get(loop.check().getId()).getIteration()); + } + + @Test + void aPersonSendsTheDraftBackTwiceThenApprovesIt() { + Block draft = llmBlockFactory.create(LLMBlockConfiguration.builder() + .name("Draft") + .llmDescriptor(ECHO) + .prompt("Revise <<${{topic}}!>>") + .build()); + Block review = humanDecisionBlockFactory.create(HumanDecisionBlockConfiguration.builder() + .name("Review") + .question("Is the draft ready?") + .options(List.of(new HumanDecisionOption("revise", "Revise"), new HumanDecisionOption("approve", "Approve"))) + .rationaleRequired(false) + .build()); + Block approved = endBlockFactory.create(EndBlockConfiguration.builder() + .name("Approved").outcomeCode("APPROVED").outcomeLabel("Approved").build()); + FlowData flow = FlowData.builder() + .block(draft).block(review).block(approved) + .connection(connection(draft, LLMBlockFactory.OUTPUT_NAME, review, HumanDecisionBlockFactory.INPUT_NAME)) + .connection(connection(review, "revise", draft, "topic")) + .connection(connection(review, "approve", approved, EndBlockFactory.INPUT_NAME)) + .build(); + + ExecutionObject execution = executionsService.createExecution("Reviewed until approved", flow); + executionsService.prepareInput(execution.getId(), draft.getId(), "topic", "idea"); + executionsService.startExecution(execution.getId()); + + for (int round = 1; round <= 2; round++) { + awaitStepWaiting(execution.getId(), review.getId(), round); + executionsService.setInteractionValue(execution.getId(), review.getId(), + HumanDecisionBlockFactory.CHOICE_FIELD, "revise"); + } + awaitStepWaiting(execution.getId(), review.getId(), 3); + + // Read back from storage in the middle of the loop: the round it is on, and the rounds + // before, have to survive the trip. + executionsService.clearInMemoryExecutions(); + ExecutionObject reloaded = executionsService.getExecution(execution.getId()); + assertEquals(3, reloaded.getContext().getSteps().get(draft.getId()).getIteration()); + assertEquals(4, reloaded.getContext().getStepHistory().size(), "two rounds of two steps each"); + executionsService.resumeExecution(execution.getId()); + + executionsService.setInteractionValue(execution.getId(), review.getId(), + HumanDecisionBlockFactory.CHOICE_FIELD, "approve"); + execution = awaitFinal(execution.getId()); + + assertEquals(ExecutionStatus.SUCCESS, execution.getContext().getStatus()); + assertEquals("idea!!!", execution.getContext().getOutcomes().getFirst().payload()); + assertEquals(List.of("APPROVED"), execution.getContext().getOutcomes().stream().map(ExecutionOutcome::code).toList()); + } + + private ExecutionObject start(RevisionLoop loop, String topic) { + ExecutionObject execution = executionsService.createExecution("Revision loop", loop.flow()); + executionsService.prepareInput(execution.getId(), loop.draft().getId(), "topic", topic); + executionsService.prepareInput(execution.getId(), loop.draft().getId(), "suffix", "x"); + this.lastExecutionId = execution.getId(); + return executionsService.startExecution(execution.getId()); + } + + private String lastExecutionId; + + private ExecutionObject executionFor(RevisionLoop loop) { + return executionsService.getExecution(lastExecutionId); + } + + private String errors(RevisionLoop loop, ExecutionObject execution) { + return execution.getContext().getErrors().toString(); + } + + /** draft -> check; the check's false output goes back to the draft's topic, its true output ends the flow. */ + private RevisionLoop revisionLoop(String condition, LoopEdgeSettings settings) { + Block draft = llmBlockFactory.create(LLMBlockConfiguration.builder() + .name("Draft") + .llmDescriptor(ECHO) + .prompt("Continue <<${{topic}}${{suffix}}>>") + .build()); + Block check = conditionalBlockFactory.create(ConditionalBlockConfiguration.builder() + .name("Long enough") + .condition(condition) + .outputTemplate("${{response}}") + .build()); + Block accepted = endBlockFactory.create(EndBlockConfiguration.builder() + .name("Accepted").outcomeCode("ACCEPTED").outcomeLabel("Accepted").build()); + Connection back = connection(check, ConditionalBlockFactory.FALSE_OUTPUT, draft, "topic"); + back.setLoop(settings); + FlowData flow = FlowData.builder() + .block(draft).block(check).block(accepted) + .connection(connection(draft, LLMBlockFactory.OUTPUT_NAME, check, "response")) + .connection(back) + .connection(connection(check, ConditionalBlockFactory.TRUE_OUTPUT, accepted, EndBlockFactory.INPUT_NAME)) + .build(); + return new RevisionLoop(flow, draft, check, accepted); + } + + private Connection connection(Block source, String sourceName, Block target, String targetName) { + return Connection.builder() + .sourceId(source.getId()).sourceName(sourceName) + .targetId(target.getId()).targetName(targetName) + .build(); + } + + private ExecutionObject awaitFinal(String executionId) { + Instant deadline = Instant.now().plus(Duration.ofSeconds(10)); + ExecutionObject execution; + do { + execution = executionsService.getExecution(executionId); + if (execution.getContext().getStatus().isFinalState()) { + return execution; + } + Thread.onSpinWait(); + } while (Instant.now().isBefore(deadline)); + throw new AssertionError("Execution did not reach a final state: " + execution.getContext().getStatus() + + " " + execution.getContext().getErrors()); + } + + private void awaitStepWaiting(String executionId, String stepId, int iteration) { + Instant deadline = Instant.now().plus(Duration.ofSeconds(10)); + do { + ExecutionObject execution = executionsService.getExecution(executionId); + var step = execution.getContext().getSteps().get(stepId); + // The step reaches its status a moment before the execution does, and an answer is only + // accepted once the execution itself is waiting. + if (step.getStatus() == StepStatus.WAITING_FOR_INTERACTION && step.getIteration() == iteration + && execution.getContext().getStatus() == ExecutionStatus.WAITING) { + return; + } + Thread.onSpinWait(); + } while (Instant.now().isBefore(deadline)); + var step = executionsService.getExecution(executionId).getContext().getSteps().get(stepId); + throw new AssertionError("Step did not wait for interaction on round " + iteration + ": " + step.getStatus() + + " round " + step.getIteration()); + } + + private record RevisionLoop(FlowData flow, Block draft, Block check, + Block accepted) { + } +} diff --git a/src/test/java/it/cnr/isti/workflow/manager/flows/loops/FlowLoopsTest.java b/src/test/java/it/cnr/isti/workflow/manager/flows/loops/FlowLoopsTest.java index a0aa79f..a71af25 100644 --- a/src/test/java/it/cnr/isti/workflow/manager/flows/loops/FlowLoopsTest.java +++ b/src/test/java/it/cnr/isti/workflow/manager/flows/loops/FlowLoopsTest.java @@ -270,6 +270,25 @@ class FlowLoopsTest { assertEquals(check.getId(), analysis.loops().getFirst().guardId()); } + @Test + void twoRoutersInARingWithNoWayInAreAmbiguousUntilTheAuthorSaysWhichWayIsBack() { + // Both routers take their first value from the person starting the run, so the shape + // alone cannot say which of them each round starts from. + Flow flow = new Flow(); + Block screen = flow.router("screen", "pass", "drop"); + Block check = flow.router("check", "again", "done"); + flow.connect(screen, "pass", check, "in"); + Connection back = flow.connect(check, "again", screen, "in"); + + assertEquals(List.of(ValidationErrorCode.LOOP_BACK_EDGE_AMBIGUOUS), codes(flow)); + + back.setLoop(new LoopEdgeSettings(null)); + FlowLoops.Analysis analysis = FlowLoops.analyze(flow.data()); + assertEquals(List.of(), analysis.problems()); + assertSame(back, analysis.loops().getFirst().backEdge()); + assertEquals(10, analysis.loops().getFirst().maxIterations(), "settings without a limit keep the default"); + } + @Test void separateLoopsAreJudgedSeparately() { Flow flow = new Flow();