lelii
/
humainflow-service
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Stop execution on branch failure and allow node renames

This commit is contained in:
Lucio Lelii 2026-03-24 11:13:10 +01:00
parent e233d27026
commit 5df9f88e38
6 changed files with 126 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionContext.java
View File

@ -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<String, Object> 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<String, Object> 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<String, Object> 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();
}
}

10
src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionObject.java
View File

@ -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();
}
}
}

1
src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionsService.java
View File

@ -320,6 +320,7 @@ public class ExecutionsService {
persist(executionObject);
cleanupManagedResourcesIfFinal(executionObject);
});
executionObject.setErrorStateListener(executionObject::abortOnError);
}
private void persist(ExecutionObject executionObject) {

6
src/main/java/it/cnr/isti/workflow/manager/flows/validation/FlowDataValidator.java
View File

@ -96,9 +96,6 @@ public class FlowDataValidator implements ConstraintValidator<ValidFlowStructure
if (!Objects.equals(block.getType().getName(), canonicalBlock.getType().getName())) {
throw validationError(error("block", block.getId(), "type", "Type does not match its configuration"));
}
if (!Objects.equals(block.getName(), canonicalBlock.getName())) {
throw validationError(error("block", block.getId(), "name", "Name does not match its configuration"));
}
if (!Objects.equals(block.getInputs(), canonicalBlock.getInputs())) {
throw validationError(error("block", block.getId(), "inputs", "Inputs do not match its configuration"));
}
@ -143,9 +140,6 @@ public class FlowDataValidator implements ConstraintValidator<ValidFlowStructure
if (!Objects.equals(container.getType().getName(), canonicalContainer.getType().getName())) {
throw validationError(error("container", container.getId(), "type", "Type does not match its configuration"));
}
if (!Objects.equals(container.getName(), canonicalContainer.getName())) {
throw validationError(error("container", container.getId(), "name", "Name does not match its configuration"));
}
if (!Objects.equals(container.getInputs(), canonicalContainer.getInputs())) {
throw validationError(error("container", container.getId(), "inputs", "Inputs do not match its configuration"));
}

40
src/test/java/it/cnr/isti/workflow/manager/controllers/FlowControllerTest.java
View File

@ -6,6 +6,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.util.List;
import java.lang.reflect.Field;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
@ -182,6 +183,35 @@ public class FlowControllerTest {
assertEquals(updatedFlowDefinition.getConnections().size(), updated.flow().getConnections().size());
}
@Test
public void createFlowAllowsRenamedBlockNodeWithoutChangingConfigurationName() {
LLMDescriptor llmDescriptor = LLMDescriptor.builder()
.provider("testProvider")
.model("testModel")
.build();
Flow flow = flowTestCreator.createFlowWithConnection(llmDescriptor);
Block<?> 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<FlowView> 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()

49
src/test/java/it/cnr/isti/workflow/manager/executions/ExecutionTest.java
View File

@ -663,6 +663,55 @@ public class ExecutionTest {
Mockito.verify(mcpAgentService, Mockito.timeout(1000)).closeSessionQuietly("shared-session-1");
}
@Test
public void failingStepAbortsOtherRunningOrWaitingBranches() {
Block<MCPAgentBlockType> failingBlock = mcpAgentBlockFactory.create(MCPAgentBlockConfiguration.builder()
.name("Failing MCP")
.model("llama3.1:8b")
.prompt("Fail for ${{candidate}}")
.mcpServers(List.of())
.build());
Block<HumanInteractionBlockType> 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();