fix(api): clear stale cancellation signals when resuming a paused workflow (#40905)

api/core/app/apps/execution_coordinator.py

@@ -8,6 +8,7 @@ from enum import Enum, auto
 
 from configs import dify_config
 from extensions.ext_redis import redis_client
+from graphon.graph_engine.command_channels import RedisChannel
 from graphon.graph_engine.manager import GraphEngineManager
 
 logger = logging.getLogger(__name__)
@@ -20,6 +21,11 @@ class AppExecutionState(Enum):
     TERMINAL = auto()
 
 
+def app_task_command_channel_key(task_id: str) -> str:
+    """Redis key of the GraphEngine command channel for one app task."""
+    return f"workflow:{task_id}:commands"
+
+
 def set_app_task_stop_flag(task_id: str) -> None:
     if not task_id:
         return
@@ -27,6 +33,40 @@ def set_app_task_stop_flag(task_id: str) -> None:
     redis_client.setex(f"generate_task_stopped:{task_id}", 600, 1)
 
 
+def clear_app_task_cancellation_signals(task_id: str) -> None:
+    """Discard cancellation signals left over from earlier attempts of one task.
+
+    Both cancellation channels are keyed by task ID and outlive the attempt that
+    armed them: the stop flag lives for 600 seconds and a queued ``AbortCommand``
+    for an hour, and neither is consumed while no engine is running. A resumed
+    workflow deliberately reuses the paused run's task ID, so without this reset
+    it inherits those signals and aborts itself as soon as it starts. Call this
+    only when starting a new attempt that is meant to run, never mid-execution.
+    """
+    if not task_id:
+        return
+
+    try:
+        redis_client.delete(f"generate_task_stopped:{task_id}")
+    except Exception:
+        logger.exception("Failed to clear stop flag for app task %s", task_id)
+
+    channel_key = app_task_command_channel_key(task_id)
+    try:
+        # fetch_commands() drains the queue and its pending marker together; the
+        # explicit delete covers a queue whose marker was already consumed.
+        discarded = RedisChannel(redis_client, channel_key).fetch_commands()
+        redis_client.delete(channel_key)
+        if discarded:
+            logger.info(
+                "Discarded %s stale GraphEngine command(s) for app task %s",
+                len(discarded),
+                task_id,
+            )
+    except Exception:
+        logger.exception("Failed to clear pending GraphEngine commands for app task %s", task_id)
+
+
 class AppExecutionCoordinator:
     """Own cancellation policy for one app execution attempt.
 

api/core/app/apps/workflow/app_runner.py

@@ -4,6 +4,7 @@ from collections.abc import Sequence
 from typing import cast
 
 from core.app.apps.base_app_queue_manager import AppQueueManager
+from core.app.apps.execution_coordinator import app_task_command_channel_key
 from core.app.apps.workflow.app_config_manager import WorkflowAppConfig
 from core.app.apps.workflow.command_channels import (
     CelerySignalCommandChannel,
@@ -153,7 +154,7 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
         # RUN WORKFLOW
         # Create Redis command channel for this workflow execution
         task_id = self.application_generate_entity.task_id
-        channel_key = f"workflow:{task_id}:commands"
+        channel_key = app_task_command_channel_key(task_id)
         celery_signal_channel = CelerySignalCommandChannel(
             shutdown_state_getter=celery_warm_shutdown_started,
             abort_reason=WORKFLOW_WARM_SHUTDOWN_ABORT_REASON,

api/tasks/app_generate/workflow_execute_task.py

@@ -12,6 +12,7 @@ from sqlalchemy import Engine, select
 from sqlalchemy.orm import Session, sessionmaker
 
 from core.app.apps.advanced_chat.app_generator import AdvancedChatAppGenerator
+from core.app.apps.execution_coordinator import clear_app_task_cancellation_signals
 from core.app.apps.message_based_app_generator import MessageBasedAppGenerator
 from core.app.apps.workflow.app_generator import WorkflowAppGenerator
 from core.app.entities.app_invoke_entities import (
@@ -558,6 +559,12 @@ def _resume_app_execution(payload: dict[str, Any]) -> None:
         )
         return
 
+    # The resumed attempt reuses the paused run's task ID, so cancellation
+    # signals armed against that ID before or during the pause would abort it
+    # immediately and report it as stopped by the user. This attempt is starting
+    # deliberately, so drop them before any engine can observe them.
+    clear_app_task_cancellation_signals(generate_entity.task_id)
+
     workflow_run_repo.resume_workflow_pause(workflow_run_id, pause_entity)
 
     pause_config = PauseStateLayerConfig(

api/tests/unit_tests/core/app/apps/test_execution_coordinator.py

@@ -2,7 +2,12 @@ from unittest.mock import Mock, patch
 
 import pytest
 
-from core.app.apps.execution_coordinator import AppExecutionCoordinator, AppExecutionState
+from core.app.apps.execution_coordinator import (
+    AppExecutionCoordinator,
+    AppExecutionState,
+    app_task_command_channel_key,
+    clear_app_task_cancellation_signals,
+)
 
 
 def test_listener_close_does_not_abort_running_attempt() -> None:
@@ -74,6 +79,54 @@ def test_pausing_started_attempt_cancels_watchdog() -> None:
     assert coordinator.state is AppExecutionState.PAUSED
 
 
+def test_clearing_cancellation_signals_drops_stop_flag_and_queued_commands() -> None:
+    channel = Mock()
+    channel.fetch_commands.return_value = [Mock()]
+    with (
+        patch("core.app.apps.execution_coordinator.redis_client") as redis_client,
+        patch("core.app.apps.execution_coordinator.RedisChannel", return_value=channel) as redis_channel,
+    ):
+        clear_app_task_cancellation_signals("task")
+
+    redis_channel.assert_called_once_with(redis_client, "workflow:task:commands")
+    channel.fetch_commands.assert_called_once_with()
+    assert redis_client.delete.call_args_list == [
+        (("generate_task_stopped:task",), {}),
+        (("workflow:task:commands",), {}),
+    ]
+
+
+def test_clearing_cancellation_signals_ignores_empty_task_id() -> None:
+    with (
+        patch("core.app.apps.execution_coordinator.redis_client") as redis_client,
+        patch("core.app.apps.execution_coordinator.RedisChannel") as redis_channel,
+    ):
+        clear_app_task_cancellation_signals("")
+
+    redis_client.delete.assert_not_called()
+    redis_channel.assert_not_called()
+
+
+def test_clearing_cancellation_signals_survives_command_channel_failure(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    with (
+        patch("core.app.apps.execution_coordinator.redis_client") as redis_client,
+        patch("core.app.apps.execution_coordinator.RedisChannel") as redis_channel,
+    ):
+        redis_channel.return_value.fetch_commands.side_effect = RuntimeError("redis read failed")
+
+        clear_app_task_cancellation_signals("task")
+
+    # The stop flag is cleared first, so a command-channel failure cannot leave it armed.
+    redis_client.delete.assert_called_once_with("generate_task_stopped:task")
+    assert "Failed to clear pending GraphEngine commands for app task task" in caplog.text
+
+
+def test_command_channel_key_matches_the_channel_the_stop_command_targets() -> None:
+    assert app_task_command_channel_key("task") == "workflow:task:commands"
+
+
 def test_stop_flag_failure_does_not_block_graph_stop(caplog: pytest.LogCaptureFixture) -> None:
     on_timeout = Mock()
     with (

api/tests/unit_tests/tasks/test_workflow_execute_task.py

@@ -852,6 +852,108 @@ def test_resume_app_execution_returns_early_when_advanced_chat_missing_conversat
     resume_advanced_chat.assert_not_called()
 
 
+def test_resume_app_execution_clears_stale_cancellation_signals_before_resuming(
+    monkeypatch: pytest.MonkeyPatch,
+    sqlite_engine: Engine,
+    sqlite_session_factory: sessionmaker[Session],
+):
+    """A resumed run reuses the paused task ID, so it must not inherit its cancellation signals.
+
+    Regression test for #40878: a stop flag or queued AbortCommand left over from
+    an earlier attempt of the same task aborted the resumed run, which then
+    finished as "Stopped by user".
+    """
+    workflow_run_id = "run-id"
+    _persist_resumption_models(sqlite_session_factory, workflow_run_id=workflow_run_id)
+
+    monkeypatch.setattr("tasks.app_generate.workflow_execute_task.db", SimpleNamespace(engine=sqlite_engine))
+
+    pause_entity = MagicMock()
+    pause_entity.get_state.return_value = b"state"
+
+    workflow_run_repo = MagicMock()
+    workflow_run_repo.get_workflow_pause.return_value = pause_entity
+    monkeypatch.setattr(
+        "tasks.app_generate.workflow_execute_task.DifyAPIRepositoryFactory.create_api_workflow_run_repository",
+        lambda *_args, **_kwargs: workflow_run_repo,
+    )
+
+    generate_entity = _build_workflow_generate_entity(stream=False)
+    resumption_context = MagicMock()
+    resumption_context.serialized_graph_runtime_state = "{}"
+    resumption_context.get_generate_entity.return_value = generate_entity
+    monkeypatch.setattr(
+        "tasks.app_generate.workflow_execute_task.WorkflowResumptionContext.loads",
+        lambda *_args, **_kwargs: resumption_context,
+    )
+    monkeypatch.setattr(
+        "tasks.app_generate.workflow_execute_task.GraphRuntimeState.from_snapshot",
+        lambda *_args, **_kwargs: MagicMock(),
+    )
+    monkeypatch.setattr(
+        "tasks.app_generate.workflow_execute_task._resolve_user_for_run", lambda *_args, **_kwargs: MagicMock()
+    )
+
+    calls: list[str] = []
+    clear_signals = MagicMock(side_effect=lambda task_id: calls.append(f"clear:{task_id}"))
+    resume_workflow = MagicMock(side_effect=lambda **_kwargs: calls.append("resume"))
+    monkeypatch.setattr("tasks.app_generate.workflow_execute_task.clear_app_task_cancellation_signals", clear_signals)
+    monkeypatch.setattr("tasks.app_generate.workflow_execute_task._resume_workflow", resume_workflow)
+
+    _resume_app_execution({"workflow_run_id": workflow_run_id})
+
+    clear_signals.assert_called_once_with(generate_entity.task_id)
+    # Clearing after the engine started would let it observe the stale abort first.
+    assert calls == [f"clear:{generate_entity.task_id}", "resume"]
+
+
+def test_resume_app_execution_keeps_cancellation_signals_when_resume_is_abandoned(
+    monkeypatch: pytest.MonkeyPatch,
+    sqlite_engine: Engine,
+    sqlite_session_factory: sessionmaker[Session],
+):
+    """No attempt is starting, so nothing may clear the signals guarding this task."""
+    workflow_run_id = "run-id"
+    _persist_resumption_models(sqlite_session_factory, workflow_run_id=workflow_run_id)
+
+    monkeypatch.setattr("tasks.app_generate.workflow_execute_task.db", SimpleNamespace(engine=sqlite_engine))
+
+    pause_entity = MagicMock()
+    pause_entity.get_state.return_value = b"state"
+
+    workflow_run_repo = MagicMock()
+    workflow_run_repo.get_workflow_pause.return_value = pause_entity
+    monkeypatch.setattr(
+        "tasks.app_generate.workflow_execute_task.DifyAPIRepositoryFactory.create_api_workflow_run_repository",
+        lambda *_args, **_kwargs: workflow_run_repo,
+    )
+
+    # Missing conversation id makes the advanced-chat resume bail out before running.
+    generate_entity = _build_advanced_chat_generate_entity(conversation_id=None)
+    resumption_context = MagicMock()
+    resumption_context.serialized_graph_runtime_state = "{}"
+    resumption_context.get_generate_entity.return_value = generate_entity
+    monkeypatch.setattr(
+        "tasks.app_generate.workflow_execute_task.WorkflowResumptionContext.loads",
+        lambda *_args, **_kwargs: resumption_context,
+    )
+    monkeypatch.setattr(
+        "tasks.app_generate.workflow_execute_task.GraphRuntimeState.from_snapshot",
+        lambda *_args, **_kwargs: MagicMock(),
+    )
+    monkeypatch.setattr(
+        "tasks.app_generate.workflow_execute_task._resolve_user_for_run", lambda *_args, **_kwargs: MagicMock()
+    )
+
+    clear_signals = MagicMock()
+    monkeypatch.setattr("tasks.app_generate.workflow_execute_task.clear_app_task_cancellation_signals", clear_signals)
+    monkeypatch.setattr("tasks.app_generate.workflow_execute_task._resume_advanced_chat", MagicMock())
+
+    _resume_app_execution({"workflow_run_id": workflow_run_id})
+
+    clear_signals.assert_not_called()
+
+
 def test_resume_advanced_chat_publishes_events_for_originally_blocking_runs(
     monkeypatch: pytest.MonkeyPatch,
     sqlite_session: Session,