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

Persist and resume executions after restart

This commit is contained in:
Lucio Lelii 2026-03-22 10:35:38 +01:00
parent 068fa695ba
commit f4dbf306b8
19 changed files with 598 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
Dockerfile
View File

@ -1,8 +1,8 @@
FROM maven:3.9.8-eclipse-temurin-21 AS build
WORKDIR /app
COPY . .
RUN mvn clean package -DskipTests
RUN ./mvnw clean package -DskipTests
FROM eclipse-temurin:21
COPY --from=build /app/target/workflow-manager.jar app.jar
COPY --from=build /app/target/humainflow.jar /app.jar
ENTRYPOINT ["java", "-jar", "/app.jar"]

2
pom.xml
View File

@ -158,7 +158,7 @@
</dependencies>
<build>
<finalName>humainFlow</finalName>
<finalName>humainflow</finalName>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>

6
src/main/java/it/cnr/isti/workflow/manager/controllers/ExecutionsController.java
View File

@ -175,6 +175,12 @@ public class ExecutionsController {
return executionService.startExecution(id);
}
@PutMapping(path = "{id}/resume")
@Operation(summary = "Resumes an execution", description = "Rebuilds a persisted execution and resumes it when possible.")
public ExecutionObject resume(@PathVariable String id) {
return executionService.resumeExecution(id);
}
@PutMapping(path = "{id}/cancel")
@Operation(summary = "Cancels an execution", description = "Cancels the execution, clears runtime state and marks it as CANCELLED.")
public ExecutionObject cancel(@PathVariable String id) {

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

@ -8,6 +8,9 @@ import java.util.Map;
import java.util.concurrent.ExecutorService;
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.steps.Input;
import it.cnr.isti.workflow.manager.executions.steps.Step;
import it.cnr.isti.workflow.manager.executions.steps.StepStatus;
@ -41,6 +44,9 @@ public class ExecutionContext implements ExecutionListener {
List<String> waitingSteps = new ArrayList<>();
@JsonIgnore
Runnable stateChangeListener;
public ExecutionContext(Map<String, Step<?>> steps) {
this.steps = steps;
this.steps.values().forEach(step -> {
@ -58,6 +64,7 @@ public class ExecutionContext implements ExecutionListener {
} else if (status == ExecutionStatus.SUCCESS || status == ExecutionStatus.ERROR || status == ExecutionStatus.CANCELLED) {
this.endTime = System.currentTimeMillis();
}
notifyStateChanged();
}
private void addResult(String nodeId, String key, Object value) {
@ -104,6 +111,7 @@ public class ExecutionContext implements ExecutionListener {
});
updateTerminalStatus();
notifyStateChanged();
}
@Override
@ -115,6 +123,7 @@ public class ExecutionContext implements ExecutionListener {
Step<?> skippedStep = this.steps.get(id);
skippedStep.getOutputs().forEach(output -> output.markUnavailable());
updateTerminalStatus();
notifyStateChanged();
}
@Override
@ -133,6 +142,7 @@ public class ExecutionContext implements ExecutionListener {
return;
}
logger.info("Step " + id + " started");
notifyStateChanged();
}
@Override
@ -149,6 +159,7 @@ public class ExecutionContext implements ExecutionListener {
this.setStatus(ExecutionStatus.WAITING);
}
}
notifyStateChanged();
}
@Override
@ -163,6 +174,7 @@ public class ExecutionContext implements ExecutionListener {
this.setStatus(ExecutionStatus.RUNNING);
}
}
notifyStateChanged();
}
@Override
@ -172,6 +184,7 @@ public class ExecutionContext implements ExecutionListener {
}
clearPartialResults(id);
partialResult.forEach((key, value) -> addPartialResult(id, key, value));
notifyStateChanged();
}
protected void setInput(String stepId, String inputName, Object value) {
@ -193,6 +206,7 @@ public class ExecutionContext implements ExecutionListener {
s.getInputs().stream().allMatch(i -> i.isRegistered() || i.isSet()))) {
this.status = ExecutionStatus.READY;
}
notifyStateChanged();
}
protected void setInteractionValue(String stepId, String fieldName, Object value) {
@ -207,11 +221,13 @@ public class ExecutionContext implements ExecutionListener {
protected void setAuthorization(String key, Object value) {
this.authorizations.put(key, value);
notifyStateChanged();
}
protected void start(ExecutorService executorService) {
this.setStatus(ExecutionStatus.RUNNING);
this.steps.values().forEach(step -> step.start(executorService));
notifyStateChanged();
}
protected synchronized void cancel() {
@ -224,6 +240,70 @@ public class ExecutionContext implements ExecutionListener {
this.waitingSteps.clear();
this.steps.values().forEach(Step::cancel);
this.setStatus(ExecutionStatus.CANCELLED);
notifyStateChanged();
}
public void setStateChangeListener(Runnable stateChangeListener) {
this.stateChangeListener = stateChangeListener;
}
public ExecutionSnapshot snapshot(Map<String, Object> providedAuthorizations) {
return ExecutionSnapshot.builder()
.status(this.status)
.startTime(this.startTime)
.endTime(this.endTime)
.providedAuthorizations(providedAuthorizations == null ? Map.of() : Map.copyOf(providedAuthorizations))
.inputs(Map.copyOf(this.inputs))
.result(Map.copyOf(this.result))
.partialResult(Map.copyOf(this.partialResult))
.errors(Map.copyOf(this.errors))
.warnings(List.copyOf(this.warnings))
.waitingSteps(List.copyOf(this.waitingSteps))
.steps(this.steps.values().stream().map(Step::snapshot).toList())
.build();
}
public void restore(ExecutionSnapshot snapshot) {
this.inputs.clear();
if (snapshot.getInputs() != null) {
this.inputs.putAll(snapshot.getInputs());
}
this.authorizations.clear();
if (snapshot.getProvidedAuthorizations() != null) {
this.authorizations.putAll(snapshot.getProvidedAuthorizations());
}
this.result.clear();
if (snapshot.getResult() != null) {
this.result.putAll(snapshot.getResult());
}
this.partialResult.clear();
if (snapshot.getPartialResult() != null) {
this.partialResult.putAll(snapshot.getPartialResult());
}
this.errors.clear();
if (snapshot.getErrors() != null) {
this.errors.putAll(snapshot.getErrors());
}
this.warnings.clear();
if (snapshot.getWarnings() != null) {
this.warnings.addAll(snapshot.getWarnings());
}
this.waitingSteps.clear();
if (snapshot.getWaitingSteps() != null) {
this.waitingSteps.addAll(snapshot.getWaitingSteps());
}
this.startTime = snapshot.getStartTime();
this.endTime = snapshot.getEndTime();
this.status = normalizeRestoredStatus(snapshot.getStatus());
}
public ExecutionStatus resumeFromSuspended() {
if (this.status != ExecutionStatus.SUSPENDED) {
return this.status;
}
ExecutionStatus resumedStatus = resolveResumableStatus();
this.setStatus(resumedStatus);
return resumedStatus;
}
private void updateTerminalStatus() {
@ -237,4 +317,30 @@ public class ExecutionContext implements ExecutionListener {
}
}
private ExecutionStatus normalizeRestoredStatus(ExecutionStatus restoredStatus) {
if (restoredStatus == null) {
return ExecutionStatus.CREATED;
}
if (restoredStatus == ExecutionStatus.RUNNING || restoredStatus == ExecutionStatus.WAITING) {
return ExecutionStatus.SUSPENDED;
}
return restoredStatus;
}
private ExecutionStatus resolveResumableStatus() {
if (!this.waitingSteps.isEmpty()) {
return ExecutionStatus.WAITING;
}
boolean allInputsSatisfied = this.steps.values().stream().allMatch(step ->
step.getStatus() == StepStatus.READY
|| step.getInputs().stream().allMatch(input -> input.isRegistered() || input.isSet()));
return allInputsSatisfied ? ExecutionStatus.READY : ExecutionStatus.CREATED;
}
private void notifyStateChanged() {
if (this.stateChangeListener != null) {
this.stateChangeListener.run();
}
}
}

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

@ -13,6 +13,8 @@ import java.util.stream.Collectors;
import com.fasterxml.jackson.annotation.JsonIgnore;
import it.cnr.isti.workflow.manager.containers.Container;
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.executors.BlockExecutors;
import it.cnr.isti.workflow.manager.executions.executors.NodeExecutors;
import it.cnr.isti.workflow.manager.executions.steps.Input;
@ -39,6 +41,9 @@ public class ExecutionObject {
String name;
@JsonIgnore
FlowData flow;
List<ExecutionAuthorizationRequirement> requiredAuthorizations = List.of();
Map<String, Object> providedAuthorizations = new HashMap<>();
@ -49,6 +54,7 @@ public class ExecutionObject {
@Builder
public ExecutionObject(String executionName, FlowData flow, List<ExecutionAuthorizationRequirement> requiredAuthorizations) {
this.name = executionName;
this.flow = flow;
this.stepConnections = flow.getConnections() == null ? List.of() : List.copyOf(flow.getConnections());
this.requiredAuthorizations = requiredAuthorizations == null ? List.of() : List.copyOf(requiredAuthorizations);
@ -117,6 +123,15 @@ public class ExecutionObject {
+ " is not in READY status (CURRENT STATUS is " + this.getContext().getStatus() + ")");
}
protected ExecutionStatus resume() {
ExecutionStatus resumedStatus = this.context.resumeFromSuspended();
if (resumedStatus == ExecutionStatus.READY) {
start();
return this.context.getStatus();
}
return resumedStatus;
}
protected void cancel() {
if (this.executorService != null) {
this.executorService.shutdownNow();
@ -148,4 +163,32 @@ public class ExecutionObject {
}
}
public void setStateChangeListener(Runnable stateChangeListener) {
this.context.setStateChangeListener(stateChangeListener);
}
public ExecutionSnapshot snapshot() {
return this.context.snapshot(this.providedAuthorizations);
}
public void restore(String executionId, long creationTime, Map<String, Object> providedAuthorizations,
ExecutionSnapshot snapshot) {
this.id = executionId;
this.creationTime = creationTime;
this.providedAuthorizations.clear();
if (providedAuthorizations != null) {
this.providedAuthorizations.putAll(providedAuthorizations);
}
if (snapshot != null && snapshot.getSteps() != null) {
Map<String, ExecutionStepSnapshot> stepsById = snapshot.getSteps().stream()
.collect(Collectors.toMap(ExecutionStepSnapshot::getStepId, Function.identity(), (left, right) -> right));
this.context.getSteps().values().forEach(step -> step.restore(stepsById.get(step.getId())));
}
if (snapshot != null) {
this.context.restore(snapshot);
}
this.providedAuthorizations.forEach(this.context::setAuthorization);
refreshInitializationStatus();
}
}

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

@ -4,6 +4,7 @@ public enum ExecutionStatus {
CREATED(true, false, false),
READY(true, false, false),
SUSPENDED(false, false, false),
RUNNING(false, false, false),
WAITING(false, false, true),
SUCCESS(false, true, false),

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

@ -17,6 +17,9 @@ import it.cnr.isti.workflow.manager.blocks.configurations.HumanInteractiveBlockC
import it.cnr.isti.workflow.manager.blocks.configurations.LLMBlockConfiguration;
import it.cnr.isti.workflow.manager.containers.Container;
import it.cnr.isti.workflow.manager.containers.configurations.ContainerConfiguration;
import it.cnr.isti.workflow.manager.executions.persistence.ExecutionSnapshot;
import it.cnr.isti.workflow.manager.executions.repo.ExecutionEntity;
import it.cnr.isti.workflow.manager.executions.repo.ExecutionRepository;
import it.cnr.isti.workflow.manager.flows.model.Flow;
import it.cnr.isti.workflow.manager.flows.model.FlowData;
import it.cnr.isti.workflow.manager.flows.validation.FlowExecutionValidator;
@ -34,6 +37,8 @@ public class ExecutionsService {
@Autowired
Map<String, LLMProvider> llmProviders;
@Autowired
ExecutionRepository executionRepository;
public ExecutionObject createExecution(String executionName, FlowData flow) {
flowExecutionValidator.validate(flow);
@ -43,7 +48,9 @@ public class ExecutionsService {
.flow(flow)
.requiredAuthorizations(requiredAuthorizations)
.build();
attachPersistence(execObject);
executions.put(execObject.getId(), execObject);
persist(execObject);
return execObject;
}
@ -64,21 +71,32 @@ public class ExecutionsService {
public ExecutionObject getExecution(String id) {
ExecutionObject toReturn = executions.get(id);
if (toReturn == null)
throw new IllegalArgumentException("Execution with id " + id + " not found");
if (toReturn == null) {
ExecutionEntity entity = executionRepository.findById(id)
.orElseThrow(() -> new IllegalArgumentException("Execution with id " + id + " not found"));
toReturn = rebuildExecution(entity);
executions.put(id, toReturn);
}
return toReturn;
}
public List<ExecutionObject> getAllExecutions() {
return executions.values().stream().toList();
return executionRepository.findAll().stream()
.map(entity -> executions.computeIfAbsent(entity.getId(), ignored -> rebuildExecution(entity)))
.toList();
}
public void removeExecution(String id) {
if (!executions.containsKey(id))
ExecutionObject execution = executions.get(id);
if (execution == null && executionRepository.existsById(id)) {
execution = getExecution(id);
}
if (execution == null)
throw new IllegalArgumentException("Execution with id " + id + " not found");
if (executions.get(id).getContext().getStatus() == ExecutionStatus.RUNNING)
if (execution.getContext().getStatus() == ExecutionStatus.RUNNING)
throw new IllegalStateException("Execution with id " + id + " is still running");
executions.remove(id);
executionRepository.deleteById(id);
}
public ExecutionObject cancelExecution(String id) {
@ -120,6 +138,25 @@ public class ExecutionsService {
return eo;
}
public ExecutionObject resumeExecution(String id) {
ExecutionObject eo = getExecution(id);
if (eo.getContext().getStatus().isFinalState()) {
return eo;
}
if (eo.getContext().getStatus() == ExecutionStatus.SUSPENDED) {
eo.resume();
return eo;
}
if (eo.getContext().getStatus() == ExecutionStatus.READY) {
eo.start();
}
return eo;
}
public void clearInMemoryExecutions() {
executions.clear();
}
private List<ExecutionAuthorizationRequirement> resolveRequiredAuthorizations(FlowData flow) {
if (flow == null || flow.getNodes().isEmpty()) {
return List.of();
@ -208,6 +245,36 @@ public class ExecutionsService {
.orElse(null);
}
private void attachPersistence(ExecutionObject executionObject) {
executionObject.setStateChangeListener(() -> persist(executionObject));
}
private void persist(ExecutionObject executionObject) {
executionRepository.save(ExecutionEntity.builder()
.id(executionObject.getId())
.name(executionObject.getName())
.creationTime(executionObject.getCreationTime())
.lastUpdateTime(System.currentTimeMillis())
.flow(executionObject.getFlow())
.snapshot(executionObject.snapshot())
.build());
}
private ExecutionObject rebuildExecution(ExecutionEntity entity) {
FlowData flow = entity.getFlow();
ExecutionObject executionObject = ExecutionObject.builder()
.executionName(entity.getName())
.flow(flow)
.requiredAuthorizations(resolveRequiredAuthorizations(flow))
.build();
ExecutionSnapshot snapshot = entity.getSnapshot();
executionObject.restore(entity.getId(), entity.getCreationTime(),
snapshot == null ? Map.of() : snapshot.getProvidedAuthorizations(),
snapshot);
attachPersistence(executionObject);
return executionObject;
}
private static class RequirementAccumulator {
private final String key;
private final String provider;

17
src/main/java/it/cnr/isti/workflow/manager/executions/FieldKey.java
View File

@ -1,6 +1,23 @@
package it.cnr.isti.workflow.manager.executions;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonValue;
public record FieldKey(String nodeId, String fieldId) {
@JsonCreator
public static FieldKey fromString(String value) {
if (value == null || value.isBlank()) {
return new FieldKey("", "");
}
int separator = value.indexOf(':');
if (separator < 0) {
return new FieldKey(value, "");
}
return new FieldKey(value.substring(0, separator), value.substring(separator + 1));
}
@JsonValue
@Override
public final String toString() {
return nodeId + ":" + fieldId;

18
src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionInputSnapshot.java Normal file
View File

@ -0,0 +1,18 @@
package it.cnr.isti.workflow.manager.executions.persistence;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class ExecutionInputSnapshot {
private String name;
private Object value;
private boolean registered;
private boolean unavailable;
}

30
src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionSnapshot.java Normal file
View File

@ -0,0 +1,30 @@
package it.cnr.isti.workflow.manager.executions.persistence;
import java.util.List;
import java.util.Map;
import it.cnr.isti.workflow.manager.executions.ExecutionStatus;
import it.cnr.isti.workflow.manager.executions.FieldKey;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class ExecutionSnapshot {
private ExecutionStatus status;
private Long startTime;
private Long endTime;
private Map<String, Object> providedAuthorizations;
private Map<FieldKey, Object> inputs;
private Map<FieldKey, Object> result;
private Map<FieldKey, Object> partialResult;
private Map<String, String> errors;
private List<String> warnings;
private List<String> waitingSteps;
private List<ExecutionStepSnapshot> steps;
}

42
src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionSnapshotConverter.java Normal file
View File

@ -0,0 +1,42 @@
package it.cnr.isti.workflow.manager.executions.persistence;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import it.cnr.isti.workflow.manager.app.ObjectMapperHolder;
import jakarta.persistence.AttributeConverter;
import jakarta.persistence.Converter;
@Converter(autoApply = false)
public class ExecutionSnapshotConverter implements AttributeConverter<ExecutionSnapshot, String> {
private static final ObjectMapper FALLBACK_MAPPER = new ObjectMapper();
@Override
public String convertToDatabaseColumn(ExecutionSnapshot snapshot) {
if (snapshot == null) {
return null;
}
try {
return mapper().writeValueAsString(snapshot);
} catch (JsonProcessingException e) {
throw new IllegalArgumentException("Errore nella serializzazione di ExecutionSnapshot in JSON", e);
}
}
@Override
public ExecutionSnapshot convertToEntityAttribute(String dbData) {
if (dbData == null || dbData.isBlank()) {
return null;
}
try {
return mapper().readValue(dbData, ExecutionSnapshot.class);
} catch (Exception e) {
throw new IllegalArgumentException("Errore nella deserializzazione di JSON in ExecutionSnapshot", e);
}
}
private ObjectMapper mapper() {
return ObjectMapperHolder.mapper != null ? ObjectMapperHolder.mapper : FALLBACK_MAPPER;
}
}

24
src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionStepSnapshot.java Normal file
View File

@ -0,0 +1,24 @@
package it.cnr.isti.workflow.manager.executions.persistence;
import java.util.List;
import java.util.Map;
import it.cnr.isti.workflow.manager.executions.steps.StepStatus;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class ExecutionStepSnapshot {
private String stepId;
private StepStatus status;
private boolean started;
private boolean simulated;
private List<ExecutionInputSnapshot> inputs;
private Map<String, Object> partialResults;
}

41
src/main/java/it/cnr/isti/workflow/manager/executions/repo/ExecutionEntity.java Normal file
View File

@ -0,0 +1,41 @@
package it.cnr.isti.workflow.manager.executions.repo;
import it.cnr.isti.workflow.manager.executions.persistence.ExecutionSnapshot;
import it.cnr.isti.workflow.manager.executions.persistence.ExecutionSnapshotConverter;
import it.cnr.isti.workflow.manager.flows.FlowConverter;
import it.cnr.isti.workflow.manager.flows.model.FlowData;
import jakarta.persistence.Column;
import jakarta.persistence.Convert;
import jakarta.persistence.Entity;
import jakarta.persistence.Id;
import jakarta.validation.constraints.NotBlank;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
@Entity
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class ExecutionEntity {
@Id
private String id;
@NotBlank
private String name;
private long creationTime;
private long lastUpdateTime;
@Column(name = "flow_data", columnDefinition = "TEXT")
@Convert(converter = FlowConverter.class)
private FlowData flow;
@Column(name = "execution_snapshot", columnDefinition = "TEXT")
@Convert(converter = ExecutionSnapshotConverter.class)
private ExecutionSnapshot snapshot;
}

6
src/main/java/it/cnr/isti/workflow/manager/executions/repo/ExecutionRepository.java Normal file
View File

@ -0,0 +1,6 @@
package it.cnr.isti.workflow.manager.executions.repo;
import org.springframework.data.jpa.repository.JpaRepository;
public interface ExecutionRepository extends JpaRepository<ExecutionEntity, String> {
}

45
src/main/java/it/cnr/isti/workflow/manager/executions/steps/Input.java
View File

@ -5,6 +5,7 @@ import java.util.List;
import com.fasterxml.jackson.annotation.JsonIgnore;
import it.cnr.isti.workflow.manager.executions.persistence.ExecutionInputSnapshot;
import it.cnr.isti.workflow.manager.ios.IODescriptor;
import it.cnr.isti.workflow.manager.ios.IOType;
import lombok.Getter;
@ -66,6 +67,21 @@ public class Input {
this.listener = listener;
}
protected ExecutionInputSnapshot snapshot() {
return ExecutionInputSnapshot.builder()
.name(descriptor.getName())
.value(value)
.registered(registered)
.unavailable(unavailable)
.build();
}
protected void restore(ExecutionInputSnapshot snapshot) {
this.value = normalizeRestoredValue(snapshot == null ? null : snapshot.getValue());
this.registered = snapshot != null && snapshot.isRegistered();
this.unavailable = snapshot != null && snapshot.isUnavailable();
}
private void validateValue(Object value) {
if (value == null) {
return;
@ -103,4 +119,33 @@ public class Input {
}
}
}
private Object normalizeRestoredValue(Object restoredValue) {
if (restoredValue == null) {
return null;
}
if (descriptor.isMultiple()) {
if (!(restoredValue instanceof List<?> values)) {
return restoredValue;
}
if (descriptor.getType() == IOType.FILE || descriptor.getType() == IOType.CSV) {
return values.stream().map(this::toFileIfNeeded).toList();
}
return restoredValue;
}
if (descriptor.getType() == IOType.FILE || descriptor.getType() == IOType.CSV) {
return toFileIfNeeded(restoredValue);
}
return restoredValue;
}
private Object toFileIfNeeded(Object value) {
if (value instanceof File) {
return value;
}
if (value instanceof String path) {
return new File(path);
}
return value;
}
}

46
src/main/java/it/cnr/isti/workflow/manager/executions/steps/Step.java
View File

@ -10,6 +10,8 @@ import com.fasterxml.jackson.annotation.JsonIgnore;
import it.cnr.isti.workflow.manager.blocks.Block;
import it.cnr.isti.workflow.manager.executions.ExecutionListener;
import it.cnr.isti.workflow.manager.executions.persistence.ExecutionInputSnapshot;
import it.cnr.isti.workflow.manager.executions.persistence.ExecutionStepSnapshot;
import it.cnr.isti.workflow.manager.executions.executors.NodeExecutors;
import it.cnr.isti.workflow.manager.flows.model.FlowNode;
import lombok.Builder;
@ -230,4 +232,48 @@ public class Step<N extends FlowNode> implements InputListener {
this.status = StepStatus.CANCELLED;
}
public ExecutionStepSnapshot snapshot() {
return ExecutionStepSnapshot.builder()
.stepId(this.id)
.status(this.status)
.started(this.started)
.simulated(this.simulated)
.inputs(this.inputs.stream().map(Input::snapshot).toList())
.partialResults(Map.copyOf(this.partialResults))
.build();
}
public void restore(ExecutionStepSnapshot snapshot) {
if (snapshot == null) {
return;
}
Map<String, ExecutionInputSnapshot> inputSnapshots = snapshot.getInputs() == null
? Map.of()
: snapshot.getInputs().stream().collect(java.util.stream.Collectors.toMap(
ExecutionInputSnapshot::getName,
java.util.function.Function.identity(),
(left, right) -> right));
this.inputs.forEach(input -> input.restore(inputSnapshots.get(input.getDescriptor().getName())));
this.partialResults.clear();
if (snapshot.getPartialResults() != null) {
this.partialResults.putAll(snapshot.getPartialResults());
}
this.started = snapshot.isStarted();
this.simulated = snapshot.isSimulated();
this.status = normalizeRestoredStatus(snapshot.getStatus());
}
private StepStatus normalizeRestoredStatus(StepStatus restoredStatus) {
if (restoredStatus == null) {
return this.status;
}
if (restoredStatus == StepStatus.RUNNING) {
if (this.node.isUserInteractive()) {
return StepStatus.WAITING_FOR_INTERACTION;
}
return this.inputs.stream().allMatch(Input::isSatisfied) ? StepStatus.READY : StepStatus.WAITING_FOR_INPUT;
}
return restoredStatus;
}
}

2
src/test/java/it/cnr/isti/workflow/manager/controllers/AssistantControllerTest.java
View File

@ -277,7 +277,7 @@ public class AssistantControllerTest {
}
private AssistantCallView waitForCallCompletion(String callId, LoginEntity user) throws Exception {
long deadline = System.currentTimeMillis() + 5_000;
long deadline = System.currentTimeMillis() + 15_000;
while (System.currentTimeMillis() < deadline) {
AssistantCallView call = assistantController.getCall(callId, user);
if (call.status() == AssistantCallStatus.COMPLETED || call.status() == AssistantCallStatus.FAILED) {

43
src/test/java/it/cnr/isti/workflow/manager/controllers/ExecutionControllerTest.java
View File

@ -22,6 +22,7 @@ import it.cnr.isti.workflow.manager.blocks.types.LLMBlockType;
import it.cnr.isti.workflow.manager.executions.ExecutionObject;
import it.cnr.isti.workflow.manager.executions.ExecutionStatus;
import it.cnr.isti.workflow.manager.executions.ExecutionAuthorizationValueRequest;
import it.cnr.isti.workflow.manager.executions.ExecutionsService;
import it.cnr.isti.workflow.manager.flows.model.Connection;
import it.cnr.isti.workflow.manager.flows.model.Flow;
import it.cnr.isti.workflow.manager.flows.model.FlowCreateRequest;
@ -44,6 +45,9 @@ public class ExecutionControllerTest {
@Autowired
private ExecutionsController executionsController;
@Autowired
private ExecutionsService executionsService;
@Test
public void createFlowAndExecution(){
LLMDescriptor llmDescriptor = LLMDescriptor.builder()
@ -256,6 +260,45 @@ public class ExecutionControllerTest {
org.junit.jupiter.api.Assertions.assertTrue(cancelled.getContext().getWarnings().isEmpty());
}
@Test
public void resumeWaitingExecutionRestoresInteractionState() {
LLMDescriptor llmDescriptor = LLMDescriptor.builder()
.provider("testProvider")
.model("testModel")
.build();
Block<HumanInteractionBlockType> reviewBlock = blocksController.create(HumanInteractiveBlockConfiguration.builder()
.name("Recruiter review")
.actionDescription("Validate candidate")
.simulateWith(llmDescriptor)
.build());
FlowCreateRequest request = new FlowCreateRequest(
"Human Resume Flow",
"Flow to test execution resume",
FlowData.builder().block(reviewBlock).build());
ResponseEntity<FlowView> createdFlow = flowController.createFlow(request, new LoginEntity("testuser", "testpassword"));
ExecutionObject executionObject = executionsController.create(createdFlow.getBody().id());
executionsController.prepareStringInputs(executionObject.getId(), reviewBlock.getId(),
reviewBlock.getInputs().getFirst().getName(), "Ada Lovelace");
executionsController.start(executionObject.getId());
waitForExecutionStatus(executionObject, ExecutionStatus.WAITING);
String executionId = executionObject.getId();
executionsService.clearInMemoryExecutions();
ExecutionObject suspended = executionsController.get(executionId);
org.junit.jupiter.api.Assertions.assertEquals(ExecutionStatus.SUSPENDED, suspended.getContext().getStatus());
ExecutionObject resumed = executionsController.resume(executionId);
org.junit.jupiter.api.Assertions.assertEquals(ExecutionStatus.WAITING, resumed.getContext().getStatus());
resumed = executionsController.provideInteractionValue(executionId, reviewBlock.getId(), "output", "Approved");
waitForExecutionStatus(resumed, ExecutionStatus.SUCCESS);
}
private void waitForExecutionStatus(ExecutionObject executionObject, ExecutionStatus expectedStatus) {
long deadline = System.currentTimeMillis() + 5_000;
while (System.currentTimeMillis() < deadline) {

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

@ -392,6 +392,60 @@ public class ExecutionTest {
Mockito.verify(mcpAgentService, Mockito.times(1)).closeSessionQuietly("session-cancel");
}
@Test
public void persistedWaitingExecutionCanBeRestoredAfterRestart() {
Block<MCPAgentChatBlockType> chatBlock = mcpAgentChatBlockFactory.create(MCPAgentChatBlockConfiguration.builder()
.name("MCP Chat")
.model("llama3.1:8b")
.inputs(List.of(new ChatInteractionInput("cand", IOType.TEXT, false)))
.mcpServers(List.of())
.build());
Flow flow = Flow.builder()
.name("Persisted MCP Chat flow")
.description("Single MCP chat block")
.block(chatBlock)
.build();
Mockito.when(mcpAgentService.openSession(Mockito.eq("llama3.1:8b"), Mockito.anyList()))
.thenReturn("session-resume");
Mockito.when(mcpAgentService.querySession("session-resume", "Hello John Doe"))
.thenReturn("MCP answer 1");
Mockito.when(mcpAgentService.querySession("session-resume", "Continue with John Doe"))
.thenReturn("MCP answer 2");
ExecutionObject execObject = executionsService.createExecution(flow);
execObject = executionsService.prepareInput(execObject.getId(), chatBlock.getId(), "cand", "John Doe");
execObject = executionsService.startExecution(execObject.getId());
while (execObject.getContext().getStatus() == ExecutionStatus.RUNNING) {
execObject = executionsService.getExecution(execObject.getId());
}
execObject = executionsService.setInteractionValue(execObject.getId(), chatBlock.getId(),
MCPAgentChatBlockFactory.INTERACTION_FIELD, "Hello ${{cand}}");
assertEquals(ExecutionStatus.WAITING, execObject.getContext().getStatus());
String executionId = execObject.getId();
executionsService.clearInMemoryExecutions();
ExecutionObject restored = executionsService.getExecution(executionId);
assertEquals(ExecutionStatus.SUSPENDED, restored.getContext().getStatus());
assertEquals("session-resume",
restored.getContext().getPartialResult().get(new FieldKey(chatBlock.getId(), "__sessionId")));
Object restoredConversation = restored.getContext().getPartialResult()
.get(new FieldKey(chatBlock.getId(), MCPAgentChatBlockFactory.HISTORY_OUTPUT));
assertTrue(restoredConversation instanceof List<?>);
assertEquals(2, ((List<?>) restoredConversation).size());
restored = executionsService.resumeExecution(executionId);
assertEquals(ExecutionStatus.WAITING, restored.getContext().getStatus());
restored = executionsService.setInteractionValue(executionId, chatBlock.getId(),
MCPAgentChatBlockFactory.INTERACTION_FIELD, "Continue with ${{cand}}");
assertEquals(ExecutionStatus.WAITING, restored.getContext().getStatus());
Mockito.verify(mcpAgentService).querySession("session-resume", "Continue with John Doe");
}
@Test
public void createInteractiveExecutionSetInputAndStart() {
ExecutionObject eo = createInteractiveExecutionAndSetInputInternally();