From b54c4ef35deb6f8ebb1e0318d1385876110b7adb Mon Sep 17 00:00:00 2001 From: Lucio Lelii Date: Fri, 25 Sep 2026 11:27:30 +0200 Subject: [PATCH] Run a flow's loops: go round, keep every round, stop at the limit When a loop's guard chooses the output that leads back, the execution resets the loop's steps and starts the next round from the entry with the value the guard produced. Every step of the loop leads to the guard, so all of them have finished by then; the reset and the delivery happen under one lock. The back edge is not wired like other connections, which push their value the moment it is produced, before the loop has been reset. Left unwired, the entry's input still counts as open when nothing else feeds it, so the first round takes it from the person starting the run. Inputs from outside the loop keep their values from round to round. The guard's other outputs are not marked as branches not taken while it goes round, since a later round may leave through them. Each step knows its round, events of loop steps carry it, and each round's steps are kept in the execution's step history, so earlier rounds and earlier verdicts stay readable after a reload. Choosing to go round with no rounds left fails the execution with LOOP_ITERATION_LIMIT_REACHED rather than leaving the loop. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../manager/executions/ExecutionContext.java | 88 ++++- .../executions/ExecutionEventType.java | 4 + .../manager/executions/ExecutionObject.java | 17 +- .../manager/executions/LoopRuntime.java | 38 ++ .../persistence/ExecutionSnapshot.java | 2 + .../persistence/ExecutionStepSnapshot.java | 2 + .../manager/executions/steps/Input.java | 10 + .../manager/executions/steps/Step.java | 68 +++- .../manager/flows/loops/FlowLoops.java | 12 +- .../manager/executions/LoopExecutionTest.java | 331 ++++++++++++++++++ .../manager/flows/loops/FlowLoopsTest.java | 19 + 11 files changed, 584 insertions(+), 7 deletions(-) create mode 100644 src/main/java/it/cnr/isti/workflow/manager/executions/LoopRuntime.java create mode 100644 src/test/java/it/cnr/isti/workflow/manager/executions/LoopExecutionTest.java diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionContext.java b/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionContext.java index 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();