From 5df9f88e388c89d20ca513ef15eebb50cb1be5d9 Mon Sep 17 00:00:00 2001 From: Lucio Lelii Date: Tue, 24 Mar 2026 11:13:10 +0100 Subject: [PATCH] Stop execution on branch failure and allow node renames --- .../manager/executions/ExecutionContext.java | 33 ++++++++++--- .../manager/executions/ExecutionObject.java | 10 ++++ .../manager/executions/ExecutionsService.java | 1 + .../flows/validation/FlowDataValidator.java | 6 --- .../controllers/FlowControllerTest.java | 40 +++++++++++++++ .../manager/executions/ExecutionTest.java | 49 +++++++++++++++++++ 6 files changed, 126 insertions(+), 13 deletions(-) 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 772688e..1ba6bb2 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 @@ -50,6 +50,9 @@ public class ExecutionContext implements ExecutionListener { @JsonIgnore Runnable stateChangeListener; + @JsonIgnore + Runnable errorStateListener; + boolean interactionSimulationEnabled = false; LLMDescriptor interactionSimulationDescriptor; @@ -112,7 +115,7 @@ public class ExecutionContext implements ExecutionListener { @Override public void completed(String id, Map result) { - if (this.status == ExecutionStatus.CANCELLED) { + if (this.status == ExecutionStatus.CANCELLED || this.status == ExecutionStatus.ERROR) { return; } logger.info("Step " + id + " completed with result: " + result); @@ -135,7 +138,7 @@ public class ExecutionContext implements ExecutionListener { @Override public void skipped(String id) { - if (this.status == ExecutionStatus.CANCELLED) { + if (this.status == ExecutionStatus.CANCELLED || this.status == ExecutionStatus.ERROR) { return; } logger.info("Step " + id + " skipped"); @@ -147,17 +150,18 @@ public class ExecutionContext implements ExecutionListener { @Override public void failed(String id, String error) { - if (this.status == ExecutionStatus.CANCELLED) { + if (this.status == ExecutionStatus.CANCELLED || this.status == ExecutionStatus.ERROR) { return; } this.addError(id, error); this.setStatus(ExecutionStatus.ERROR); + abortAfterFailure(id); logger.severe("Step " + id + " failed with error: " + error); } @Override public synchronized void started(String id) { - if (this.status == ExecutionStatus.CANCELLED) { + if (this.status == ExecutionStatus.CANCELLED || this.status == ExecutionStatus.ERROR) { return; } logger.info("Step " + id + " started"); @@ -166,7 +170,7 @@ public class ExecutionContext implements ExecutionListener { @Override public synchronized void paused(String id) { - if (this.status == ExecutionStatus.CANCELLED) { + if (this.status == ExecutionStatus.CANCELLED || this.status == ExecutionStatus.ERROR) { return; } logger.info("Step " + id + " paused"); @@ -183,7 +187,7 @@ public class ExecutionContext implements ExecutionListener { @Override public synchronized void resumed(String id) { - if (this.status == ExecutionStatus.CANCELLED) { + if (this.status == ExecutionStatus.CANCELLED || this.status == ExecutionStatus.ERROR) { return; } logger.info("Step " + id + " resumed"); @@ -198,7 +202,7 @@ public class ExecutionContext implements ExecutionListener { @Override public synchronized void partialUpdated(String id, Map partialResult) { - if (this.status == ExecutionStatus.CANCELLED) { + if (this.status == ExecutionStatus.CANCELLED || this.status == ExecutionStatus.ERROR) { return; } clearPartialResults(id); @@ -317,6 +321,10 @@ public class ExecutionContext implements ExecutionListener { this.stateChangeListener = stateChangeListener; } + public void setErrorStateListener(Runnable errorStateListener) { + this.errorStateListener = errorStateListener; + } + public ExecutionSnapshot snapshot(Map providedAuthorizations) { return ExecutionSnapshot.builder() .status(this.status) @@ -440,4 +448,15 @@ public class ExecutionContext implements ExecutionListener { } } + private synchronized void abortAfterFailure(String failedStepId) { + this.waitingSteps.clear(); + this.steps.values().stream() + .filter(step -> !step.getId().equals(failedStepId)) + .forEach(Step::cancel); + if (this.errorStateListener != null) { + this.errorStateListener.run(); + } + notifyStateChanged(); + } + } 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 c6fdb4d..44c908f 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 @@ -227,6 +227,10 @@ public class ExecutionObject { this.context.setStateChangeListener(stateChangeListener); } + public void setErrorStateListener(Runnable errorStateListener) { + this.context.setErrorStateListener(errorStateListener); + } + public ExecutionSnapshot snapshot() { return this.context.snapshot(this.providedAuthorizations); } @@ -269,4 +273,10 @@ public class ExecutionObject { return hasInteractiveSteps; } + protected void abortOnError() { + if (this.executorService != null) { + this.executorService.shutdownNow(); + } + } + } 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 ee29c51..d37b523 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 @@ -320,6 +320,7 @@ public class ExecutionsService { persist(executionObject); cleanupManagedResourcesIfFinal(executionObject); }); + executionObject.setErrorStateListener(executionObject::abortOnError); } private void persist(ExecutionObject executionObject) { diff --git a/src/main/java/it/cnr/isti/workflow/manager/flows/validation/FlowDataValidator.java b/src/main/java/it/cnr/isti/workflow/manager/flows/validation/FlowDataValidator.java index 13601be..39d6eb4 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/flows/validation/FlowDataValidator.java +++ b/src/main/java/it/cnr/isti/workflow/manager/flows/validation/FlowDataValidator.java @@ -96,9 +96,6 @@ public class FlowDataValidator implements ConstraintValidator renamedBlock = flow.getBlocks().getFirst(); + overwriteField(renamedBlock, "name", "Renamed in editor"); + + FlowCreateRequest request = new FlowCreateRequest( + flow.getName(), + flow.getDescription(), + FlowData.builder() + .blocks(flow.getBlocks()) + .connections(flow.getConnections()) + .build()); + + ResponseEntity createResponse = flowController.createFlow( + request, + new LoginEntity("testuser", "testpassword")); + + assertTrue(createResponse.getStatusCode().is2xxSuccessful()); + assertNotNull(createResponse.getBody()); + assertEquals(FlowViewStatus.EXECUTABLE, createResponse.getBody().status()); + assertEquals("Renamed in editor", createResponse.getBody().flow().getBlocks().getFirst().getName()); + } + @Test public void updateMissingFlowReturnsNotFound() { LLMDescriptor llmDescriptor = LLMDescriptor.builder() @@ -238,6 +268,16 @@ public class FlowControllerTest { .noneMatch(f -> created.id().equals(f.id()))); } + private void overwriteField(Object target, String fieldName, Object value) { + try { + Field field = target.getClass().getDeclaredField(fieldName); + field.setAccessible(true); + field.set(target, value); + } catch (ReflectiveOperationException e) { + throw new RuntimeException(e); + } + } + @Test public void deleteFlowReturnsForbiddenForDifferentOwner() { LLMDescriptor llmDescriptor = LLMDescriptor.builder() diff --git a/src/test/java/it/cnr/isti/workflow/manager/executions/ExecutionTest.java b/src/test/java/it/cnr/isti/workflow/manager/executions/ExecutionTest.java index 84c0f80..00648a2 100644 --- a/src/test/java/it/cnr/isti/workflow/manager/executions/ExecutionTest.java +++ b/src/test/java/it/cnr/isti/workflow/manager/executions/ExecutionTest.java @@ -663,6 +663,55 @@ public class ExecutionTest { Mockito.verify(mcpAgentService, Mockito.timeout(1000)).closeSessionQuietly("shared-session-1"); } + @Test + public void failingStepAbortsOtherRunningOrWaitingBranches() { + Block failingBlock = mcpAgentBlockFactory.create(MCPAgentBlockConfiguration.builder() + .name("Failing MCP") + .model("llama3.1:8b") + .prompt("Fail for ${{candidate}}") + .mcpServers(List.of()) + .build()); + + Block humanBlock = humanInteractiveBlockFactory.create( + HumanInteractiveBlockConfiguration.builder() + .name("Human branch") + .actionDescription("Review ${{input}}") + .build()); + + Flow flow = Flow.builder() + .name("Parallel failure flow") + .description("A failing branch should abort the execution") + .block(failingBlock) + .block(humanBlock) + .build(); + + Mockito.when(mcpAgentService.execute("llama3.1:8b", "Fail for Ada", List.of())) + .thenThrow(new IllegalStateException("MCP exploded")); + + ExecutionObject execObject = executionsService.createExecution(flow); + execObject = executionsService.prepareInput(execObject.getId(), failingBlock.getId(), "candidate", "Ada"); + execObject = executionsService.prepareInput(execObject.getId(), humanBlock.getId(), "input", "Ada"); + execObject = executionsService.startExecution(execObject.getId()); + + while (execObject.getContext().getStatus() == ExecutionStatus.RUNNING + || execObject.getContext().getStatus() == ExecutionStatus.WAITING) { + try { + Thread.sleep(50); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + execObject = executionsService.getExecution(execObject.getId()); + } + + assertEquals(ExecutionStatus.ERROR, execObject.getContext().getStatus()); + assertTrue(execObject.getContext().getErrors().containsKey(failingBlock.getId())); + Step humanStep = execObject.getContext().getSteps().get(humanBlock.getId()); + assertNotNull(humanStep); + assertEquals(StepStatus.CANCELLED, humanStep.getStatus()); + assertTrue(execObject.getContext().getWaitingSteps().isEmpty()); + } + @Test public void createInteractiveExecutionSetInputAndStart() { ExecutionObject eo = createInteractiveExecutionAndSetInputInternally();