fix(agent): bound streams and cancel zombie workflow runs (#39186)

api/core/app/apps/agent_app/app_generator.py

@@ -540,6 +540,9 @@ class AgentAppGenerator(MessageBasedAppGenerator):
                 base_url=dify_config.AGENT_BACKEND_BASE_URL,
                 use_fake=dify_config.AGENT_BACKEND_USE_FAKE,
                 fake_scenario=dify_config.AGENT_BACKEND_FAKE_SCENARIO,
+                stream_read_timeout_seconds=dify_config.AGENT_BACKEND_STREAM_READ_TIMEOUT_SECONDS,
+                stream_max_reconnects=dify_config.AGENT_BACKEND_STREAM_MAX_RECONNECTS,
+                stream_run_timeout_seconds=dify_config.AGENT_BACKEND_RUN_TIMEOUT_SECONDS,
             ),
             event_adapter=AgentBackendRunEventAdapter(),
             session_store=AgentAppRuntimeSessionStore(),

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

@@ -941,48 +941,64 @@ class AgentAppRunner:
             if pending_text:
                 persist_answer_text(pending_text)
 
-        for public_event in self._agent_backend_client.stream_events(run_id):
-            if queue_manager.is_stopped():
-                flush_pending_agent_message_text()
-                self._cancel_run(run_id)
-                raise GenerateTaskStoppedError()
-            for internal_event in self._event_adapter.adapt(public_event):
+        try:
+            public_events = self._agent_backend_client.stream_events(
+                run_id,
+                should_stop=queue_manager.is_stopped,
+            )
+            for public_event in public_events:
                 if queue_manager.is_stopped():
                     flush_pending_agent_message_text()
                     self._cancel_run(run_id)
                     raise GenerateTaskStoppedError()
-                if internal_event.type in (
-                    AgentBackendInternalEventType.RUN_STARTED,
-                    AgentBackendInternalEventType.STREAM_EVENT,
-                    AgentBackendInternalEventType.AGENT_MESSAGE_DELTA,
-                ):
-                    if isinstance(internal_event, AgentBackendAgentMessageDeltaInternalEvent):
-                        debounced_delta = text_delta_debouncer.push(internal_event.delta)
-                        if debounced_delta:
-                            persist_answer_text(debounced_delta)
-                        continue
-
-                    if isinstance(internal_event, AgentBackendStreamInternalEvent):
+                for internal_event in self._event_adapter.adapt(public_event):
+                    if queue_manager.is_stopped():
                         flush_pending_agent_message_text()
-                        try:
-                            process_recorder.handle_stream_event(internal_event)
-                        except Exception:
-                            db.session.rollback()
-                            logger.warning(
-                                "Failed to persist Agent App process event: run_id=%s message_id=%s event_kind=%s",
-                                run_id,
-                                message_id,
-                                internal_event.event_kind,
-                                exc_info=True,
-                            )
+                        self._cancel_run(run_id)
+                        raise GenerateTaskStoppedError()
+                    if internal_event.type in (
+                        AgentBackendInternalEventType.RUN_STARTED,
+                        AgentBackendInternalEventType.STREAM_EVENT,
+                        AgentBackendInternalEventType.AGENT_MESSAGE_DELTA,
+                    ):
+                        if isinstance(internal_event, AgentBackendAgentMessageDeltaInternalEvent):
+                            debounced_delta = text_delta_debouncer.push(internal_event.delta)
+                            if debounced_delta:
+                                persist_answer_text(debounced_delta)
+                            continue
+
+                        if isinstance(internal_event, AgentBackendStreamInternalEvent):
+                            flush_pending_agent_message_text()
+                            try:
+                                process_recorder.handle_stream_event(internal_event)
+                            except Exception:
+                                db.session.rollback()
+                                logger.warning(
+                                    "Failed to persist Agent App process event: run_id=%s message_id=%s event_kind=%s",
+                                    run_id,
+                                    message_id,
+                                    internal_event.event_kind,
+                                    exc_info=True,
+                                )
+                            continue
                         continue
-                    continue
-                flush_pending_agent_message_text()
-                terminal = internal_event
-                break
-            if terminal is not None:
-                break
+                    flush_pending_agent_message_text()
+                    terminal = internal_event
+                    break
+                if terminal is not None:
+                    break
+        except GenerateTaskStoppedError:
+            raise
+        except Exception as error:
+            flush_pending_agent_message_text()
+            self._cancel_run(run_id)
+            if queue_manager.is_stopped():
+                raise GenerateTaskStoppedError() from error
+            raise
         flush_pending_agent_message_text()
+        if queue_manager.is_stopped():
+            self._cancel_run(run_id)
+            raise GenerateTaskStoppedError()
         return terminal, process_recorder
 
     def _cancel_run(self, run_id: str) -> None:

api/core/app/apps/base_app_queue_manager.py

@@ -21,6 +21,7 @@ from core.app.entities.queue_entities import (
     WorkflowQueueMessage,
 )
 from extensions.ext_redis import redis_client
+from graphon.graph_engine.manager import GraphEngineManager
 from graphon.runtime import GraphRuntimeState
 
 logger = logging.getLogger(__name__)
@@ -51,6 +52,9 @@ class AppQueueManager(ABC):
         self._graph_runtime_state: GraphRuntimeState | None = None
         self._stopped_cache: TTLCache[tuple, bool] = TTLCache(maxsize=1, ttl=1)
         self._cache_lock = threading.Lock()
+        self._execution_terminal = threading.Event()
+        self._abort_sent = threading.Event()
+        self._lifecycle_lock = threading.Lock()
 
     def listen(self):
         """
@@ -59,7 +63,7 @@ class AppQueueManager(ABC):
         """
         # wait for APP_MAX_EXECUTION_TIME seconds to stop listen
         listen_timeout = dify_config.APP_MAX_EXECUTION_TIME
-        start_time = time.time()
+        start_time = time.monotonic()
         last_ping_time: int | float = 0
         try:
             while True:
@@ -72,8 +76,14 @@ class AppQueueManager(ABC):
                 except queue.Empty:
                     continue
                 finally:
-                    elapsed_time = time.time() - start_time
-                    if elapsed_time >= listen_timeout or self._is_stopped():
+                    elapsed_time = time.monotonic() - start_time
+                    timed_out = elapsed_time >= listen_timeout
+                    manually_stopped = self._is_stopped()
+                    if not self._execution_terminal.is_set() and (timed_out or manually_stopped):
+                        reason = (
+                            f"App execution exceeded {listen_timeout} seconds" if timed_out else "App task was stopped"
+                        )
+                        self._abort_execution(reason)
                         # publish two messages to make sure the client can receive the stop signal
                         # and stop listening after the stop signal processed
                         self.publish(
@@ -84,16 +94,33 @@ class AppQueueManager(ABC):
                         self.publish(QueuePingEvent(), PublishFrom.TASK_PIPELINE)
                         last_ping_time = elapsed_time // 10
         finally:
+            if not self._execution_terminal.is_set():
+                self._abort_execution("Client response stream closed before app execution completed")
             self._graph_runtime_state = None  # Release reference once consumers finish or close the generator.
 
-    def stop_listen(self):
+    def stop_listen(self, *, execution_terminal: bool = False):
         """
         Stop listen to queue
         :return:
         """
+        if execution_terminal:
+            self._execution_terminal.set()
         self._clear_task_belong_cache()
         self._q.put(None)
 
+    def _abort_execution(self, reason: str) -> None:
+        """Propagate response timeout/disconnect to legacy and GraphEngine runners."""
+        with self._lifecycle_lock:
+            if self._execution_terminal.is_set() or self._abort_sent.is_set():
+                return
+            self._abort_sent.set()
+
+        try:
+            self.set_stop_flag_no_user_check(self._task_id)
+            GraphEngineManager(redis_client).send_stop_command(self._task_id, reason=reason)
+        except Exception:
+            logger.exception("Failed to abort app execution for task %s", self._task_id)
+
     def _clear_task_belong_cache(self) -> None:
         """
         Remove the task belong cache key once listening is finished.

api/core/app/apps/message_based_app_queue_manager.py

@@ -45,7 +45,7 @@ class MessageBasedAppQueueManager(AppQueueManager):
         if isinstance(
             event, QueueStopEvent | QueueErrorEvent | QueueMessageEndEvent | QueueAdvancedChatMessageEndEvent
         ):
-            self.stop_listen()
+            self.stop_listen(execution_terminal=True)
 
         if pub_from == PublishFrom.APPLICATION_MANAGER and self._is_stopped():
             if self._app_mode == AppMode.ADVANCED_CHAT.value:

api/core/app/apps/pipeline/pipeline_queue_manager.py

@@ -42,7 +42,7 @@ class PipelineQueueManager(AppQueueManager):
             | QueueWorkflowFailedEvent
             | QueueWorkflowPartialSuccessEvent,
         ):
-            self.stop_listen()
+            self.stop_listen(execution_terminal=True)
 
         if pub_from == PublishFrom.APPLICATION_MANAGER and self._is_stopped():
             raise GenerateTaskStoppedError()

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

@@ -41,4 +41,4 @@ class WorkflowAppQueueManager(AppQueueManager):
             | QueueWorkflowFailedEvent
             | QueueWorkflowPartialSuccessEvent,
         ):
-            self.stop_listen()
+            self.stop_listen(execution_terminal=True)

api/core/app/apps/workflow_app_runner.py

@@ -26,6 +26,7 @@ from core.app.entities.queue_entities import (
     QueueNodeSucceededEvent,
     QueueReasoningChunkEvent,
     QueueRetrieverResourcesEvent,
+    QueueStopEvent,
     QueueTextChunkEvent,
     QueueWorkflowFailedEvent,
     QueueWorkflowPartialSuccessEvent,
@@ -424,7 +425,12 @@ class WorkflowBasedAppRunner:
                     QueueWorkflowFailedEvent(error=event.error, exceptions_count=event.exceptions_count)
                 )
             case GraphRunAbortedEvent():
-                self._publish_event(QueueWorkflowFailedEvent(error=event.reason or "Unknown error", exceptions_count=0))
+                self._publish_event(
+                    QueueStopEvent(
+                        stopped_by=QueueStopEvent.StopBy.USER_MANUAL,
+                        reason=event.reason or "Workflow execution aborted",
+                    )
+                )
             case GraphRunPausedEvent():
                 runtime_state = workflow_entry.graph_engine.graph_runtime_state
                 paused_nodes = runtime_state.get_paused_nodes()

api/core/app/entities/queue_entities.py

@@ -500,11 +500,15 @@ class QueueStopEvent(AppQueueEvent):
 
     event: QueueEvent = QueueEvent.STOP
     stopped_by: StopBy
+    reason: str | None = None
 
     def get_stop_reason(self) -> str:
         """
         To stop reason
         """
+        if self.reason:
+            return self.reason
+
         reason_mapping = {
             QueueStopEvent.StopBy.USER_MANUAL: "Stopped by user.",
             QueueStopEvent.StopBy.ANNOTATION_REPLY: "Stopped by annotation reply.",

api/core/workflow/node_factory.py

@@ -499,6 +499,9 @@ class DifyNodeFactory(NodeFactory):
                     base_url=dify_config.AGENT_BACKEND_BASE_URL,
                     use_fake=dify_config.AGENT_BACKEND_USE_FAKE,
                     fake_scenario=dify_config.AGENT_BACKEND_FAKE_SCENARIO,
+                    stream_read_timeout_seconds=dify_config.AGENT_BACKEND_STREAM_READ_TIMEOUT_SECONDS,
+                    stream_max_reconnects=dify_config.AGENT_BACKEND_STREAM_MAX_RECONNECTS,
+                    stream_run_timeout_seconds=dify_config.AGENT_BACKEND_RUN_TIMEOUT_SECONDS,
                 ),
                 "event_adapter": AgentBackendRunEventAdapter(),
                 # Agent Files §4.6: reback file outputs from the ToolFile row so

api/core/workflow/nodes/agent_v2/agent_node.py

@@ -5,6 +5,7 @@ from collections.abc import Generator, Mapping, Sequence
 from typing import TYPE_CHECKING, Any, override
 
 from agenton.compositor import CompositorSessionSnapshot
+from dify_agent.protocol import CancelRunRequest
 
 from clients.agent_backend import (
     AgentBackendAgentMessageDeltaInternalEvent,
@@ -473,7 +474,10 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
         """
         stream_event_count = 0
         try:
-            for public_event in self._agent_backend_client.stream_events(run_id):
+            for public_event in self._agent_backend_client.stream_events(
+                run_id,
+                should_stop=self._is_graph_aborted,
+            ):
                 stream_event_count += 1
                 for internal_event in self._event_adapter.adapt(public_event):
                     if internal_event.type == AgentBackendInternalEventType.RUN_STARTED:
@@ -501,6 +505,7 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
                         | AgentBackendDeferredToolCallInternalEvent,
                     ):
                         return internal_event, None
+                    self._cancel_backend_run(run_id, reason="unexpected_event")
                     return None, self._failure_event(
                         inputs={},
                         process_data={},
@@ -509,6 +514,7 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
                         error_type="agent_backend_stream_error",
                     )
         except AgentBackendError as error:
+            self._cancel_backend_run(run_id, reason=self._stream_stop_reason())
             return None, self._failure_event(
                 inputs={},
                 process_data={},
@@ -517,6 +523,7 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
                 error_type=self._agent_backend_error_type(error),
             )
         except Exception as error:
+            self._cancel_backend_run(run_id, reason=self._stream_stop_reason())
             return None, self._failure_event(
                 inputs={},
                 process_data={},
@@ -525,8 +532,28 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
                 error_type="agent_backend_stream_error",
             )
 
+        self._cancel_backend_run(run_id, reason="stream_ended_without_terminal_event")
         return None, None
 
+    def _is_graph_aborted(self) -> bool:
+        """Let Agent SSE consumption observe GraphEngine's cooperative abort state."""
+        try:
+            return self.graph_runtime_state.graph_execution.aborted
+        except (AttributeError, RuntimeError):
+            return False
+
+    def _stream_stop_reason(self) -> str:
+        return "workflow_graph_aborted" if self._is_graph_aborted() else "event_stream_failed"
+
+    def _cancel_backend_run(self, run_id: str, *, reason: str) -> None:
+        try:
+            self._agent_backend_client.cancel_run(
+                run_id,
+                CancelRunRequest(reason=reason, message="Workflow Agent event consumption stopped"),
+            )
+        except Exception:
+            logger.warning("Failed to cancel Workflow Agent backend run: run_id=%s", run_id, exc_info=True)
+
     @staticmethod
     def _record_type_check_metadata(metadata: dict[str, Any], outcome: OutputTypeCheckOutcome) -> None:
         # Surface enough detail in metadata for Inspector / debug logs without