diff --git a/Dockerfile b/Dockerfile index f4c4a19..3474028 100644 --- a/Dockerfile +++ b/Dockerfile @@ -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"] diff --git a/pom.xml b/pom.xml index 9839292..934c266 100644 --- a/pom.xml +++ b/pom.xml @@ -158,7 +158,7 @@ - humainFlow + humainflow org.apache.maven.plugins diff --git a/src/main/java/it/cnr/isti/workflow/manager/controllers/ExecutionsController.java b/src/main/java/it/cnr/isti/workflow/manager/controllers/ExecutionsController.java index ff96961..e3783d6 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/controllers/ExecutionsController.java +++ b/src/main/java/it/cnr/isti/workflow/manager/controllers/ExecutionsController.java @@ -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) { 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 0186957..0edea90 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 @@ -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 waitingSteps = new ArrayList<>(); + @JsonIgnore + Runnable stateChangeListener; + public ExecutionContext(Map> 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 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(); + } + } + } 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 2760988..17685b6 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 @@ -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 requiredAuthorizations = List.of(); Map providedAuthorizations = new HashMap<>(); @@ -49,6 +54,7 @@ public class ExecutionObject { @Builder public ExecutionObject(String executionName, FlowData flow, List 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 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 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(); + } + } diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionStatus.java b/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionStatus.java index 86ef6fc..97aa3d5 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionStatus.java +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/ExecutionStatus.java @@ -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), 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 bb5a901..2a70217 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 @@ -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 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 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 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; diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/FieldKey.java b/src/main/java/it/cnr/isti/workflow/manager/executions/FieldKey.java index 160e87f..b87d712 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/executions/FieldKey.java +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/FieldKey.java @@ -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; diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionInputSnapshot.java b/src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionInputSnapshot.java new file mode 100644 index 0000000..775dbd6 --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionInputSnapshot.java @@ -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; +} 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 new file mode 100644 index 0000000..4db2b3c --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionSnapshot.java @@ -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 providedAuthorizations; + private Map inputs; + private Map result; + private Map partialResult; + private Map errors; + private List warnings; + private List waitingSteps; + private List steps; +} diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionSnapshotConverter.java b/src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionSnapshotConverter.java new file mode 100644 index 0000000..4278a2f --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionSnapshotConverter.java @@ -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 { + + 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; + } +} 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 new file mode 100644 index 0000000..e246a5d --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/persistence/ExecutionStepSnapshot.java @@ -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 inputs; + private Map partialResults; +} diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/repo/ExecutionEntity.java b/src/main/java/it/cnr/isti/workflow/manager/executions/repo/ExecutionEntity.java new file mode 100644 index 0000000..4e87dee --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/repo/ExecutionEntity.java @@ -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; +} diff --git a/src/main/java/it/cnr/isti/workflow/manager/executions/repo/ExecutionRepository.java b/src/main/java/it/cnr/isti/workflow/manager/executions/repo/ExecutionRepository.java new file mode 100644 index 0000000..1b0fa7b --- /dev/null +++ b/src/main/java/it/cnr/isti/workflow/manager/executions/repo/ExecutionRepository.java @@ -0,0 +1,6 @@ +package it.cnr.isti.workflow.manager.executions.repo; + +import org.springframework.data.jpa.repository.JpaRepository; + +public interface ExecutionRepository extends JpaRepository { +} 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 2161095..11471f2 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 @@ -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; + } } 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 0e01c54..a577c89 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,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 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 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; + } + } diff --git a/src/test/java/it/cnr/isti/workflow/manager/controllers/AssistantControllerTest.java b/src/test/java/it/cnr/isti/workflow/manager/controllers/AssistantControllerTest.java index c4d4e05..9dc8ad2 100644 --- a/src/test/java/it/cnr/isti/workflow/manager/controllers/AssistantControllerTest.java +++ b/src/test/java/it/cnr/isti/workflow/manager/controllers/AssistantControllerTest.java @@ -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) { diff --git a/src/test/java/it/cnr/isti/workflow/manager/controllers/ExecutionControllerTest.java b/src/test/java/it/cnr/isti/workflow/manager/controllers/ExecutionControllerTest.java index 311c6e4..ff612ef 100644 --- a/src/test/java/it/cnr/isti/workflow/manager/controllers/ExecutionControllerTest.java +++ b/src/test/java/it/cnr/isti/workflow/manager/controllers/ExecutionControllerTest.java @@ -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 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 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) { 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 99bc660..b8fc465 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 @@ -392,6 +392,60 @@ public class ExecutionTest { Mockito.verify(mcpAgentService, Mockito.times(1)).closeSessionQuietly("session-cancel"); } + @Test + public void persistedWaitingExecutionCanBeRestoredAfterRestart() { + Block 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();