Add workflow cancellation support for Redis queue mode

src/workflow/CommandExecutor.py

@@ -4,12 +4,19 @@
 import subprocess
 import threading
 from pathlib import Path
+from typing import Callable, Optional
 from .Logger import Logger
 from .ParameterManager import ParameterManager
 import sys
 import importlib.util
 import json
 
+
+class WorkflowCancelled(Exception):
+    """Raised when a workflow is cancelled by the user."""
+    pass
+
+
 class CommandExecutor:
     """
     Manages the execution of external shell commands such as OpenMS TOPP tools within a Streamlit application.
@@ -24,6 +31,27 @@ def __init__(self, workflow_dir: Path, logger: Logger, parameter_manager: Parame
         self.pid_dir = Path(workflow_dir, "pids")
         self.logger = logger
         self.parameter_manager = parameter_manager
+        self._should_stop: Optional[Callable[[], bool]] = None
+
+    def set_cancellation_check(self, should_stop_func: Callable[[], bool]) -> None:
+        """
+        Set a function that checks if the workflow should be cancelled.
+
+        Args:
+            should_stop_func: A callable that returns True if the workflow should stop.
+        """
+        self._should_stop = should_stop_func
+
+    def _check_cancellation(self) -> None:
+        """
+        Check if the workflow was cancelled and raise WorkflowCancelled if so.
+
+        This is called before each command execution to allow stopping workflows
+        between commands when running in Redis queue mode.
+        """
+        if self._should_stop and self._should_stop():
+            self.logger.log("Workflow cancelled by user")
+            raise WorkflowCancelled("Workflow was cancelled")
 
     def run_multiple_commands(
         self, commands: list[str]
@@ -81,8 +109,12 @@ def run_command(self, command: list[str]) -> bool:
             command (list[str]): The shell command to execute, provided as a list of strings.
 
         Raises:
+            WorkflowCancelled: If the workflow was cancelled by the user.
             Exception: If the command execution results in any errors.
         """
+        # Check for cancellation before starting the command
+        self._check_cancellation()
+
         # Ensure all command parts are strings
         command = [str(c) for c in command]
 

src/workflow/tasks.py

@@ -11,6 +11,8 @@
 import traceback
 from pathlib import Path
 
+from .CommandExecutor import WorkflowCancelled
+
 
 def execute_workflow(
     workflow_dir: str,
@@ -74,6 +76,16 @@ def execute_workflow(
         executor = CommandExecutor(workflow_path, logger, parameter_manager)
         executor.pid_dir.mkdir(parents=True, exist_ok=True)
 
+        # Set up cancellation check for Redis queue mode
+        if job is not None:
+            def should_stop():
+                try:
+                    job.refresh()  # Get latest status from Redis
+                    return job.is_stopped
+                except Exception:
+                    return False
+            executor.set_cancellation_check(should_stop)
+
         _update_progress(job, 0.1, "Starting workflow execution...")
 
         # Create workflow instance
@@ -118,6 +130,32 @@ def execute_workflow(
             "message": "Workflow completed successfully"
         }
 
+    except WorkflowCancelled:
+        # Handle user-initiated cancellation cleanly
+        try:
+            log_dir = workflow_path / "logs"
+            log_dir.mkdir(parents=True, exist_ok=True)
+            for log_name in ["minimal.log", "commands-and-run-times.log", "all.log"]:
+                log_file = log_dir / log_name
+                with open(log_file, "a") as f:
+                    f.write("\n\nWorkflow cancelled by user\n")
+        except Exception:
+            pass
+
+        # Clean up pid directory
+        try:
+            pid_dir = workflow_path / "pids"
+            shutil.rmtree(pid_dir, ignore_errors=True)
+        except Exception:
+            pass
+
+        return {
+            "success": False,
+            "workflow_dir": str(workflow_path),
+            "cancelled": True,
+            "error": "Workflow cancelled by user"
+        }
+
     except Exception as e:
         error_msg = f"Workflow failed: {str(e)}\n{traceback.format_exc()}"