Merge pull request #384 from OpenMS/claude/fix-vendor-queue-error-EwPar

fix(queue): show "Workflow was cancelled" on first Stop click

Author: t0mdavid-m

src/workflow/StreamlitUI.py

@@ -20,6 +20,7 @@
     tk_directory_dialog,
     tk_file_dialog,
 )
+from src.workflow._log_status import classify_log_outcome
 
 
 class StreamlitUI:
@@ -1268,9 +1269,11 @@ def execution_section(
             with open(log_path, "r", encoding="utf-8") as f:
                 lines = f.readlines()
             content = "".join(lines)
-            # Check if workflow finished successfully
-            if "WORKFLOW FINISHED" in content:
+            outcome = classify_log_outcome(content)
+            if outcome == "finished":
                 st.success("**Workflow completed successfully.**")
+            elif outcome == "cancelled":
+                st.warning("**Workflow was cancelled.**")
             else:
                 st.error("**Errors occurred, check log file.**")
             # Apply line limit to static display
@@ -1324,6 +1327,9 @@ def _show_queue_status(self, status: dict) -> None:
                 with st.expander("Error Details", expanded=True):
                     st.code(job_error)
 
+        elif job_status == "canceled":
+            st.warning(f"**Status: {label}** - Workflow was cancelled.")
+
         # Expandable job details
         with st.expander("Job Details", expanded=False):
             st.code(f"""Job ID: {status.get('job_id', 'N/A')}

src/workflow/WorkflowManager.py

@@ -178,16 +178,28 @@ def stop_workflow(self) -> bool:
         """
         Stop a running workflow.
 
+        Writes a "WORKFLOW CANCELLED" marker to the log so the static
+        run-page display can render a "Workflow was cancelled" message
+        instead of "Errors occurred". Cleans up the worker-side pid_dir
+        left behind when the RQ worker is force-stopped, so a subsequent
+        get_workflow_status does not flip running back to True via the
+        local-mode fallback.
+
+        .job_id is intentionally left in place: get_job_info will report
+        the canceled status to the UI so _show_queue_status can render the
+        Cancelled pill. Resubmission overwrites it; RQ's result_ttl
+        eventually evicts the job and get_workflow_status self-heals.
+
         Returns:
-            True if successfully stopped
+            True if a stop action was taken (queue cancel or local kill).
         """
         # Try to cancel queue job first (online mode)
         if self._queue_manager and self._queue_manager.is_available:
             job_id = self._queue_manager.load_job_id(self.workflow_dir)
             if job_id:
-                success = self._queue_manager.cancel_job(job_id)
-                if success:
-                    self._queue_manager.clear_job_id(self.workflow_dir)
+                if self._queue_manager.cancel_job(job_id):
+                    self.logger.log("WORKFLOW CANCELLED")
+                    shutil.rmtree(self.executor.pid_dir, ignore_errors=True)
                     return True
 
         # Fallback: stop local process
@@ -214,6 +226,8 @@ def _stop_local_workflow(self) -> bool:
 
         # Clean up the pid directory
         shutil.rmtree(pid_dir, ignore_errors=True)
+        if stopped:
+            self.logger.log("WORKFLOW CANCELLED")
         return stopped
 
     def show_file_upload_section(self) -> None:

src/workflow/_log_status.py

@@ -0,0 +1,30 @@
+"""
+Pure helper for classifying a workflow log file's terminal state.
+
+Kept streamlit-free so the static-display dispatch in StreamlitUI can be
+unit-tested without a Streamlit runtime.
+"""
+
+from typing import Literal
+
+LogOutcome = Literal["finished", "cancelled", "error"]
+
+CANCELLED_MARKER = "WORKFLOW CANCELLED"
+FINISHED_MARKER = "WORKFLOW FINISHED"
+
+
+def classify_log_outcome(content: str) -> LogOutcome:
+    """
+    Classify a workflow log's terminal state from its full text.
+
+    Order matters: a TOPP subprocess often dies as the worker is being torn
+    down, so a partial 'ERROR:' line followed by the cancellation marker is
+    still a cancellation, not a crash. Cancellation therefore wins over
+    finished (defensive — both shouldn't appear) and over the implicit error
+    fallback.
+    """
+    if CANCELLED_MARKER in content:
+        return "cancelled"
+    if FINISHED_MARKER in content:
+        return "finished"
+    return "error"

tests/test_log_status.py

@@ -0,0 +1,50 @@
+"""
+Tests for the classify_log_outcome helper used by the run-page static display.
+
+The UI must render three different messages depending on what's in the
+workflow log:
+    finished  -> "Workflow completed successfully" (success)
+    cancelled -> "Workflow was cancelled"          (warning)
+    error     -> "Errors occurred, check log file" (error)
+
+This helper is split out so the dispatch is unit-testable without booting
+Streamlit and without pulling in pyopenms.
+"""
+
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from src.workflow._log_status import classify_log_outcome
+
+
+def test_finished_marker_returns_finished():
+    assert classify_log_outcome(
+        "STARTING WORKFLOW\n\nstep 1\n\nWORKFLOW FINISHED\n\n"
+    ) == "finished"
+
+
+def test_cancelled_marker_returns_cancelled():
+    assert classify_log_outcome(
+        "STARTING WORKFLOW\n\nstep 1\n\nWORKFLOW CANCELLED\n\n"
+    ) == "cancelled"
+
+
+def test_cancelled_takes_precedence_over_partial_error():
+    """
+    A TOPP subprocess often dies as the worker is being torn down, leaving
+    an ERROR line followed by the cancellation marker. The user-meaningful
+    state is 'cancelled', not 'error'.
+    """
+    assert classify_log_outcome(
+        "STARTING WORKFLOW\n\nERROR: subprocess died\n\nWORKFLOW CANCELLED\n\n"
+    ) == "cancelled"
+
+
+def test_truncated_log_returns_error():
+    assert classify_log_outcome("STARTING WORKFLOW\n\nstep 1\n\n") == "error"
+
+
+def test_empty_log_returns_error():
+    assert classify_log_outcome("") == "error"

tests/test_workflow_manager_stop.py

@@ -0,0 +1,162 @@
+"""
+Tests for WorkflowManager.stop_workflow / get_workflow_status interaction.
+
+Bug being fixed (follow-up to PR #383): the RQ worker actually terminates when
+the user clicks "Stop Workflow", but the Streamlit UI still shows
+"workflow is running" and pressing Stop a second time produces an
+"error has occurred" message instead of "workflow has been cancelled".
+
+Root causes:
+  1. stop_workflow clears .job_id on success, so the next get_workflow_status
+     poll falls through to the local-mode pid_dir fallback. The killed worker
+     left stale child PID files in pid_dir, so the fallback wrongly returns
+     running=True.
+  2. The worker never wrote 'WORKFLOW FINISHED' to the log because it was
+     killed mid-execution. The UI's static-display branch only knows two
+     outcomes (FINISHED -> success, else -> error), so a cancelled run is
+     misreported as an error.
+
+These tests pin both behaviours.
+"""
+
+import os
+import sys
+import types
+
+import pytest
+
+fakeredis = pytest.importorskip("fakeredis")
+rq = pytest.importorskip("rq")
+streamlit = pytest.importorskip("streamlit")
+pyopenms = pytest.importorskip("pyopenms")
+
+from rq import Queue
+from rq.job import Job, JobStatus
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from src.workflow.Logger import Logger
+from src.workflow.QueueManager import QueueManager
+from src.workflow.WorkflowManager import WorkflowManager
+
+
+def _make_queue_manager() -> QueueManager:
+    """Build a QueueManager wired to fake Redis, bypassing __init__."""
+    qm = QueueManager.__new__(QueueManager)
+    qm._redis = fakeredis.FakeStrictRedis()
+    qm._queue = Queue(QueueManager.QUEUE_NAME, connection=qm._redis)
+    qm._is_online = True
+    qm._init_attempted = True
+    qm._default_timeout = 7200
+    qm._default_result_ttl = 86400
+    return qm
+
+
+def _force_started(job: Job, worker_name: str = "rq:worker:test-worker") -> None:
+    job.set_status(JobStatus.STARTED)
+    job.worker_name = worker_name
+    job.save()
+
+
+def _make_workflow_manager(tmp_path, monkeypatch) -> WorkflowManager:
+    """
+    Build a minimal WorkflowManager wired to a fakeredis-backed QueueManager.
+
+    Bypasses __init__ (which constructs a StreamlitUI and reads session
+    state). Only the attributes used by stop_workflow / get_workflow_status
+    are populated:
+        - workflow_dir  (real tmp dir)
+        - logger        (real Logger; streamlit-free)
+        - executor      (SimpleNamespace exposing pid_dir; CommandExecutor
+                         itself imports streamlit so we cannot instantiate it)
+        - _queue_manager (fakeredis-backed QueueManager)
+
+    A stale child PID file is dropped in pid_dir to simulate the state the
+    worker leaves behind when it is force-killed mid-execution.
+    """
+    workflow_dir = tmp_path / "wf"
+    workflow_dir.mkdir()
+
+    pid_dir = workflow_dir / "pids"
+    pid_dir.mkdir()
+    (pid_dir / "12345").touch()
+
+    qm = _make_queue_manager()
+    job = qm._queue.enqueue(os.getcwd, job_id="wf-job")
+    _force_started(job)
+    qm.store_job_id(workflow_dir, "wf-job")
+
+    monkeypatch.setattr(
+        "rq.command.send_stop_job_command",
+        lambda *a, **kw: None,
+    )
+
+    wm = WorkflowManager.__new__(WorkflowManager)
+    wm.workflow_dir = workflow_dir
+    wm.logger = Logger(workflow_dir)
+    wm.executor = types.SimpleNamespace(pid_dir=pid_dir)
+    wm._queue_manager = qm
+    return wm
+
+
+def test_stop_workflow_clears_running_state_in_queue_mode(tmp_path, monkeypatch):
+    """
+    Bug #1: after a successful queue cancel, get_workflow_status must report
+    running=False. Currently the stale pid_dir keeps the local-mode fallback
+    returning running=True.
+    """
+    wm = _make_workflow_manager(tmp_path, monkeypatch)
+
+    assert wm.stop_workflow() is True
+
+    status = wm.get_workflow_status()
+    assert status["running"] is False, (
+        "After cancel, get_workflow_status must not report the workflow as "
+        "still running."
+    )
+
+    pid_dir = wm.executor.pid_dir
+    assert not (pid_dir.exists() and any(pid_dir.iterdir())), (
+        "stop_workflow must clean up the stale pid_dir left behind by the "
+        "killed worker; otherwise the local-mode fallback in "
+        "get_workflow_status flips running back to True."
+    )
+
+
+def test_stop_workflow_writes_cancellation_marker_to_log(tmp_path, monkeypatch):
+    """
+    Bug #2: the static log-display branch needs a way to tell 'cancelled'
+    apart from 'crashed'. stop_workflow must drop a 'WORKFLOW CANCELLED'
+    marker into the log so the UI can render the right message.
+    """
+    wm = _make_workflow_manager(tmp_path, monkeypatch)
+    wm.logger.log("STARTING WORKFLOW")  # mimic a partial run
+
+    assert wm.stop_workflow() is True
+
+    logs_dir = wm.workflow_dir / "logs"
+    for log_name in ("minimal.log", "commands-and-run-times.log", "all.log"):
+        content = (logs_dir / log_name).read_text(encoding="utf-8")
+        assert "WORKFLOW CANCELLED" in content, (
+            f"{log_name} should contain the WORKFLOW CANCELLED marker."
+        )
+        assert "WORKFLOW FINISHED" not in content, (
+            f"{log_name} must not claim the workflow finished."
+        )
+
+
+def test_stop_workflow_is_idempotent(tmp_path, monkeypatch):
+    """
+    Pressing Stop a second time (or stop firing twice on Streamlit rerun)
+    must not raise and must keep running=False. The first call's user intent
+    has already been satisfied; subsequent calls should be safe no-ops.
+    """
+    wm = _make_workflow_manager(tmp_path, monkeypatch)
+
+    assert wm.stop_workflow() is True
+
+    # Second call: must not raise, get_workflow_status must remain not-running.
+    wm.stop_workflow()
+    assert wm.get_workflow_status()["running"] is False
+
+