feat: unified management stop event (#30479)

api/core/workflow/graph_engine/graph_engine.py

@@ -8,6 +8,7 @@ Domain-Driven Design principles for improved maintainability and testability.
 import contextvars
 import logging
 import queue
+import threading
 from collections.abc import Generator
 from typing import TYPE_CHECKING, cast, final
 
@@ -75,10 +76,13 @@ class GraphEngine:
         scale_down_idle_time: float | None = None,
     ) -> None:
         """Initialize the graph engine with all subsystems and dependencies."""
+        # stop event
+        self._stop_event = threading.Event()
 
         # Bind runtime state to current workflow context
         self._graph = graph
         self._graph_runtime_state = graph_runtime_state
+        self._graph_runtime_state.stop_event = self._stop_event
         self._graph_runtime_state.configure(graph=cast("GraphProtocol", graph))
         self._command_channel = command_channel
 
@@ -177,6 +181,7 @@ class GraphEngine:
             max_workers=self._max_workers,
             scale_up_threshold=self._scale_up_threshold,
             scale_down_idle_time=self._scale_down_idle_time,
+            stop_event=self._stop_event,
         )
 
         # === Orchestration ===
@@ -207,6 +212,7 @@ class GraphEngine:
             event_handler=self._event_handler_registry,
             execution_coordinator=self._execution_coordinator,
             event_emitter=self._event_manager,
+            stop_event=self._stop_event,
         )
 
         # === Validation ===
@@ -324,6 +330,7 @@ class GraphEngine:
 
     def _start_execution(self, *, resume: bool = False) -> None:
         """Start execution subsystems."""
+        self._stop_event.clear()
         paused_nodes: list[str] = []
         if resume:
             paused_nodes = self._graph_runtime_state.consume_paused_nodes()
@@ -351,13 +358,12 @@ class GraphEngine:
 
     def _stop_execution(self) -> None:
         """Stop execution subsystems."""
+        self._stop_event.set()
         self._dispatcher.stop()
         self._worker_pool.stop()
         # Don't mark complete here as the dispatcher already does it
 
         # Notify layers
-        logger = logging.getLogger(__name__)
-
         for layer in self._layers:
             try:
                 layer.on_graph_end(self._graph_execution.error)

api/core/workflow/graph_engine/orchestration/dispatcher.py

@@ -44,6 +44,7 @@ class Dispatcher:
         event_queue: queue.Queue[GraphNodeEventBase],
         event_handler: "EventHandler",
         execution_coordinator: ExecutionCoordinator,
+        stop_event: threading.Event,
         event_emitter: EventManager | None = None,
     ) -> None:
         """
@@ -61,7 +62,7 @@ class Dispatcher:
         self._event_emitter = event_emitter
 
         self._thread: threading.Thread | None = None
-        self._stop_event = threading.Event()
+        self._stop_event = stop_event
         self._start_time: float | None = None
 
     def start(self) -> None:
@@ -69,16 +70,14 @@ class Dispatcher:
         if self._thread and self._thread.is_alive():
             return
 
-        self._stop_event.clear()
         self._start_time = time.time()
         self._thread = threading.Thread(target=self._dispatcher_loop, name="GraphDispatcher", daemon=True)
         self._thread.start()
 
     def stop(self) -> None:
         """Stop the dispatcher thread."""
-        self._stop_event.set()
         if self._thread and self._thread.is_alive():
-            self._thread.join(timeout=10.0)
+            self._thread.join(timeout=2.0)
 
     def _dispatcher_loop(self) -> None:
         """Main dispatcher loop."""

api/core/workflow/graph_engine/worker.py

@@ -42,6 +42,7 @@ class Worker(threading.Thread):
         event_queue: queue.Queue[GraphNodeEventBase],
         graph: Graph,
         layers: Sequence[GraphEngineLayer],
+        stop_event: threading.Event,
         worker_id: int = 0,
         flask_app: Flask | None = None,
         context_vars: contextvars.Context | None = None,
@@ -65,13 +66,16 @@ class Worker(threading.Thread):
         self._worker_id = worker_id
         self._flask_app = flask_app
         self._context_vars = context_vars
-        self._stop_event = threading.Event()
         self._last_task_time = time.time()
+        self._stop_event = stop_event
         self._layers = layers if layers is not None else []
 
     def stop(self) -> None:
-        """Signal the worker to stop processing."""
-        self._stop_event.set()
+        """Worker is controlled via shared stop_event from GraphEngine.
+
+        This method is a no-op retained for backward compatibility.
+        """
+        pass
 
     @property
     def is_idle(self) -> bool:

api/core/workflow/graph_engine/worker_management/worker_pool.py

@@ -41,6 +41,7 @@ class WorkerPool:
         event_queue: queue.Queue[GraphNodeEventBase],
         graph: Graph,
         layers: list[GraphEngineLayer],
+        stop_event: threading.Event,
         flask_app: "Flask | None" = None,
         context_vars: "Context | None" = None,
         min_workers: int | None = None,
@@ -81,6 +82,7 @@ class WorkerPool:
         self._worker_counter = 0
         self._lock = threading.RLock()
         self._running = False
+        self._stop_event = stop_event
 
         # No longer tracking worker states with callbacks to avoid lock contention
 
@@ -135,7 +137,7 @@ class WorkerPool:
             # Wait for workers to finish
             for worker in self._workers:
                 if worker.is_alive():
-                    worker.join(timeout=10.0)
+                    worker.join(timeout=2.0)
 
             self._workers.clear()
 
@@ -152,6 +154,7 @@ class WorkerPool:
             worker_id=worker_id,
             flask_app=self._flask_app,
             context_vars=self._context_vars,
+            stop_event=self._stop_event,
         )
 
         worker.start()

api/core/workflow/nodes/base/node.py

@@ -264,6 +264,10 @@ class Node(Generic[NodeDataT]):
         """
         raise NotImplementedError
 
+    def _should_stop(self) -> bool:
+        """Check if execution should be stopped."""
+        return self.graph_runtime_state.stop_event.is_set()
+
     def run(self) -> Generator[GraphNodeEventBase, None, None]:
         execution_id = self.ensure_execution_id()
         self._start_at = naive_utc_now()
@@ -332,6 +336,21 @@ class Node(Generic[NodeDataT]):
                     yield event
                 else:
                     yield event
+
+                if self._should_stop():
+                    error_message = "Execution cancelled"
+                    yield NodeRunFailedEvent(
+                        id=self.execution_id,
+                        node_id=self._node_id,
+                        node_type=self.node_type,
+                        start_at=self._start_at,
+                        node_run_result=NodeRunResult(
+                            status=WorkflowNodeExecutionStatus.FAILED,
+                            error=error_message,
+                        ),
+                        error=error_message,
+                    )
+                    return
         except Exception as e:
             logger.exception("Node %s failed to run", self._node_id)
             result = NodeRunResult(

api/core/workflow/runtime/graph_runtime_state.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import importlib
 import json
+import threading
 from collections.abc import Mapping, Sequence
 from copy import deepcopy
 from dataclasses import dataclass
@@ -168,6 +169,7 @@ class GraphRuntimeState:
         self._pending_response_coordinator_dump: str | None = None
         self._pending_graph_execution_workflow_id: str | None = None
         self._paused_nodes: set[str] = set()
+        self.stop_event: threading.Event = threading.Event()
 
         if graph is not None:
             self.attach_graph(graph)

api/tests/unit_tests/core/workflow/graph_engine/orchestration/test_dispatcher.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import queue
+import threading
 from unittest import mock
 
 from core.workflow.entities.pause_reason import SchedulingPause
@@ -36,6 +37,7 @@ def test_dispatcher_should_consume_remains_events_after_pause():
         event_queue=event_queue,
         event_handler=event_handler,
         execution_coordinator=execution_coordinator,
+        stop_event=threading.Event(),
     )
     dispatcher._dispatcher_loop()
     assert event_queue.empty()
@@ -96,6 +98,7 @@ def _run_dispatcher_for_event(event) -> int:
         event_queue=event_queue,
         event_handler=event_handler,
         execution_coordinator=coordinator,
+        stop_event=threading.Event(),
     )
 
     dispatcher._dispatcher_loop()
@@ -181,6 +184,7 @@ def test_dispatcher_drain_event_queue():
         event_queue=event_queue,
         event_handler=event_handler,
         execution_coordinator=coordinator,
+        stop_event=threading.Event(),
     )
 
     dispatcher._dispatcher_loop()

api/tests/unit_tests/core/workflow/graph_engine/test_stop_event.py

@@ -0,0 +1,539 @@
+"""
+Unit tests for stop_event functionality in GraphEngine.
+
+Tests the unified stop_event management by GraphEngine and its propagation
+to WorkerPool, Worker, Dispatcher, and Nodes.
+"""
+
+import threading
+import time
+from unittest.mock import MagicMock, Mock, patch
+
+from core.app.entities.app_invoke_entities import InvokeFrom
+from core.workflow.entities.graph_init_params import GraphInitParams
+from core.workflow.graph import Graph
+from core.workflow.graph_engine import GraphEngine
+from core.workflow.graph_engine.command_channels import InMemoryChannel
+from core.workflow.graph_events import (
+    GraphRunStartedEvent,
+    GraphRunSucceededEvent,
+    NodeRunStartedEvent,
+)
+from core.workflow.nodes.answer.answer_node import AnswerNode
+from core.workflow.nodes.start.start_node import StartNode
+from core.workflow.runtime import GraphRuntimeState, VariablePool
+from models.enums import UserFrom
+
+
+class TestStopEventPropagation:
+    """Test suite for stop_event propagation through GraphEngine components."""
+
+    def test_graph_engine_creates_stop_event(self):
+        """Test that GraphEngine creates a stop_event on initialization."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+        mock_graph = MagicMock(spec=Graph)
+        mock_graph.nodes = {}
+        mock_graph.edges = {}
+        mock_graph.root_node = MagicMock()
+
+        engine = GraphEngine(
+            workflow_id="test_workflow",
+            graph=mock_graph,
+            graph_runtime_state=runtime_state,
+            command_channel=InMemoryChannel(),
+        )
+
+        # Verify stop_event was created
+        assert engine._stop_event is not None
+        assert isinstance(engine._stop_event, threading.Event)
+
+        # Verify it was set in graph_runtime_state
+        assert runtime_state.stop_event is not None
+        assert runtime_state.stop_event is engine._stop_event
+
+    def test_stop_event_cleared_on_start(self):
+        """Test that stop_event is cleared when execution starts."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+        mock_graph = MagicMock(spec=Graph)
+        mock_graph.nodes = {}
+        mock_graph.edges = {}
+        mock_graph.root_node = MagicMock()
+        mock_graph.root_node.id = "start"  # Set proper id
+
+        start_node = StartNode(
+            id="start",
+            config={"id": "start", "data": {"title": "start", "variables": []}},
+            graph_init_params=GraphInitParams(
+                tenant_id="test_tenant",
+                app_id="test_app",
+                workflow_id="test_workflow",
+                graph_config={},
+                user_id="test_user",
+                user_from=UserFrom.ACCOUNT,
+                invoke_from=InvokeFrom.DEBUGGER,
+                call_depth=0,
+            ),
+            graph_runtime_state=runtime_state,
+        )
+        mock_graph.nodes["start"] = start_node
+        mock_graph.get_outgoing_edges = MagicMock(return_value=[])
+        mock_graph.get_incoming_edges = MagicMock(return_value=[])
+
+        engine = GraphEngine(
+            workflow_id="test_workflow",
+            graph=mock_graph,
+            graph_runtime_state=runtime_state,
+            command_channel=InMemoryChannel(),
+        )
+
+        # Set the stop_event before running
+        engine._stop_event.set()
+        assert engine._stop_event.is_set()
+
+        # Run the engine (should clear the stop_event)
+        events = list(engine.run())
+
+        # After running, stop_event should be set again (by _stop_execution)
+        # But during start it was cleared
+        assert any(isinstance(e, GraphRunStartedEvent) for e in events)
+        assert any(isinstance(e, GraphRunSucceededEvent) for e in events)
+
+    def test_stop_event_set_on_stop(self):
+        """Test that stop_event is set when execution stops."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+        mock_graph = MagicMock(spec=Graph)
+        mock_graph.nodes = {}
+        mock_graph.edges = {}
+        mock_graph.root_node = MagicMock()
+        mock_graph.root_node.id = "start"  # Set proper id
+
+        start_node = StartNode(
+            id="start",
+            config={"id": "start", "data": {"title": "start", "variables": []}},
+            graph_init_params=GraphInitParams(
+                tenant_id="test_tenant",
+                app_id="test_app",
+                workflow_id="test_workflow",
+                graph_config={},
+                user_id="test_user",
+                user_from=UserFrom.ACCOUNT,
+                invoke_from=InvokeFrom.DEBUGGER,
+                call_depth=0,
+            ),
+            graph_runtime_state=runtime_state,
+        )
+        mock_graph.nodes["start"] = start_node
+        mock_graph.get_outgoing_edges = MagicMock(return_value=[])
+        mock_graph.get_incoming_edges = MagicMock(return_value=[])
+
+        engine = GraphEngine(
+            workflow_id="test_workflow",
+            graph=mock_graph,
+            graph_runtime_state=runtime_state,
+            command_channel=InMemoryChannel(),
+        )
+
+        # Initially not set
+        assert not engine._stop_event.is_set()
+
+        # Run the engine
+        list(engine.run())
+
+        # After execution completes, stop_event should be set
+        assert engine._stop_event.is_set()
+
+    def test_stop_event_passed_to_worker_pool(self):
+        """Test that stop_event is passed to WorkerPool."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+        mock_graph = MagicMock(spec=Graph)
+        mock_graph.nodes = {}
+        mock_graph.edges = {}
+        mock_graph.root_node = MagicMock()
+
+        engine = GraphEngine(
+            workflow_id="test_workflow",
+            graph=mock_graph,
+            graph_runtime_state=runtime_state,
+            command_channel=InMemoryChannel(),
+        )
+
+        # Verify WorkerPool has the stop_event
+        assert engine._worker_pool._stop_event is not None
+        assert engine._worker_pool._stop_event is engine._stop_event
+
+    def test_stop_event_passed_to_dispatcher(self):
+        """Test that stop_event is passed to Dispatcher."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+        mock_graph = MagicMock(spec=Graph)
+        mock_graph.nodes = {}
+        mock_graph.edges = {}
+        mock_graph.root_node = MagicMock()
+
+        engine = GraphEngine(
+            workflow_id="test_workflow",
+            graph=mock_graph,
+            graph_runtime_state=runtime_state,
+            command_channel=InMemoryChannel(),
+        )
+
+        # Verify Dispatcher has the stop_event
+        assert engine._dispatcher._stop_event is not None
+        assert engine._dispatcher._stop_event is engine._stop_event
+
+
+class TestNodeStopCheck:
+    """Test suite for Node._should_stop() functionality."""
+
+    def test_node_should_stop_checks_runtime_state(self):
+        """Test that Node._should_stop() checks GraphRuntimeState.stop_event."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+
+        answer_node = AnswerNode(
+            id="answer",
+            config={"id": "answer", "data": {"title": "answer", "answer": "{{#start.result#}}"}},
+            graph_init_params=GraphInitParams(
+                tenant_id="test_tenant",
+                app_id="test_app",
+                workflow_id="test_workflow",
+                graph_config={},
+                user_id="test_user",
+                user_from=UserFrom.ACCOUNT,
+                invoke_from=InvokeFrom.DEBUGGER,
+                call_depth=0,
+            ),
+            graph_runtime_state=runtime_state,
+        )
+
+        # Initially stop_event is not set
+        assert not answer_node._should_stop()
+
+        # Set the stop_event
+        runtime_state.stop_event.set()
+
+        # Now _should_stop should return True
+        assert answer_node._should_stop()
+
+    def test_node_run_checks_stop_event_between_yields(self):
+        """Test that Node.run() checks stop_event between yielding events."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+
+        # Create a simple node
+        answer_node = AnswerNode(
+            id="answer",
+            config={"id": "answer", "data": {"title": "answer", "answer": "hello"}},
+            graph_init_params=GraphInitParams(
+                tenant_id="test_tenant",
+                app_id="test_app",
+                workflow_id="test_workflow",
+                graph_config={},
+                user_id="test_user",
+                user_from=UserFrom.ACCOUNT,
+                invoke_from=InvokeFrom.DEBUGGER,
+                call_depth=0,
+            ),
+            graph_runtime_state=runtime_state,
+        )
+
+        # Set stop_event BEFORE running the node
+        runtime_state.stop_event.set()
+
+        # Run the node - should yield start event then detect stop
+        # The node should check stop_event before processing
+        assert answer_node._should_stop(), "stop_event should be set"
+
+        # Run and collect events
+        events = list(answer_node.run())
+
+        # Since stop_event is set at the start, we should get:
+        # 1. NodeRunStartedEvent (always yielded first)
+        # 2. Either NodeRunFailedEvent (if detected early) or NodeRunSucceededEvent (if too fast)
+        assert len(events) >= 2
+        assert isinstance(events[0], NodeRunStartedEvent)
+
+        # Note: AnswerNode is very simple and might complete before stop check
+        # The important thing is that _should_stop() returns True when stop_event is set
+        assert answer_node._should_stop()
+
+
+class TestStopEventIntegration:
+    """Integration tests for stop_event in workflow execution."""
+
+    def test_simple_workflow_respects_stop_event(self):
+        """Test that a simple workflow respects stop_event."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+
+        mock_graph = MagicMock(spec=Graph)
+        mock_graph.nodes = {}
+        mock_graph.edges = {}
+        mock_graph.root_node = MagicMock()
+        mock_graph.root_node.id = "start"
+
+        # Create start and answer nodes
+        start_node = StartNode(
+            id="start",
+            config={"id": "start", "data": {"title": "start", "variables": []}},
+            graph_init_params=GraphInitParams(
+                tenant_id="test_tenant",
+                app_id="test_app",
+                workflow_id="test_workflow",
+                graph_config={},
+                user_id="test_user",
+                user_from=UserFrom.ACCOUNT,
+                invoke_from=InvokeFrom.DEBUGGER,
+                call_depth=0,
+            ),
+            graph_runtime_state=runtime_state,
+        )
+
+        answer_node = AnswerNode(
+            id="answer",
+            config={"id": "answer", "data": {"title": "answer", "answer": "hello"}},
+            graph_init_params=GraphInitParams(
+                tenant_id="test_tenant",
+                app_id="test_app",
+                workflow_id="test_workflow",
+                graph_config={},
+                user_id="test_user",
+                user_from=UserFrom.ACCOUNT,
+                invoke_from=InvokeFrom.DEBUGGER,
+                call_depth=0,
+            ),
+            graph_runtime_state=runtime_state,
+        )
+
+        mock_graph.nodes["start"] = start_node
+        mock_graph.nodes["answer"] = answer_node
+        mock_graph.get_outgoing_edges = MagicMock(return_value=[])
+        mock_graph.get_incoming_edges = MagicMock(return_value=[])
+
+        engine = GraphEngine(
+            workflow_id="test_workflow",
+            graph=mock_graph,
+            graph_runtime_state=runtime_state,
+            command_channel=InMemoryChannel(),
+        )
+
+        # Set stop_event before running
+        runtime_state.stop_event.set()
+
+        # Run the engine
+        events = list(engine.run())
+
+        # Should get started event but not succeeded (due to stop)
+        assert any(isinstance(e, GraphRunStartedEvent) for e in events)
+        # The workflow should still complete (start node runs quickly)
+        # but answer node might be cancelled depending on timing
+
+    def test_stop_event_with_concurrent_nodes(self):
+        """Test stop_event behavior with multiple concurrent nodes."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+
+        mock_graph = MagicMock(spec=Graph)
+        mock_graph.nodes = {}
+        mock_graph.edges = {}
+        mock_graph.root_node = MagicMock()
+
+        # Create multiple nodes
+        for i in range(3):
+            answer_node = AnswerNode(
+                id=f"answer_{i}",
+                config={"id": f"answer_{i}", "data": {"title": f"answer_{i}", "answer": f"test{i}"}},
+                graph_init_params=GraphInitParams(
+                    tenant_id="test_tenant",
+                    app_id="test_app",
+                    workflow_id="test_workflow",
+                    graph_config={},
+                    user_id="test_user",
+                    user_from=UserFrom.ACCOUNT,
+                    invoke_from=InvokeFrom.DEBUGGER,
+                    call_depth=0,
+                ),
+                graph_runtime_state=runtime_state,
+            )
+            mock_graph.nodes[f"answer_{i}"] = answer_node
+
+        mock_graph.get_outgoing_edges = MagicMock(return_value=[])
+        mock_graph.get_incoming_edges = MagicMock(return_value=[])
+
+        engine = GraphEngine(
+            workflow_id="test_workflow",
+            graph=mock_graph,
+            graph_runtime_state=runtime_state,
+            command_channel=InMemoryChannel(),
+        )
+
+        # All nodes should share the same stop_event
+        for node in mock_graph.nodes.values():
+            assert node.graph_runtime_state.stop_event is runtime_state.stop_event
+            assert node.graph_runtime_state.stop_event is engine._stop_event
+
+
+class TestStopEventTimeoutBehavior:
+    """Test stop_event behavior with join timeouts."""
+
+    @patch("core.workflow.graph_engine.orchestration.dispatcher.threading.Thread")
+    def test_dispatcher_uses_shorter_timeout(self, mock_thread_cls: MagicMock):
+        """Test that Dispatcher uses 2s timeout instead of 10s."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+        mock_graph = MagicMock(spec=Graph)
+        mock_graph.nodes = {}
+        mock_graph.edges = {}
+        mock_graph.root_node = MagicMock()
+
+        engine = GraphEngine(
+            workflow_id="test_workflow",
+            graph=mock_graph,
+            graph_runtime_state=runtime_state,
+            command_channel=InMemoryChannel(),
+        )
+
+        dispatcher = engine._dispatcher
+        dispatcher.start()  # This will create and start the mocked thread
+
+        mock_thread_instance = mock_thread_cls.return_value
+        mock_thread_instance.is_alive.return_value = True
+
+        dispatcher.stop()
+
+        mock_thread_instance.join.assert_called_once_with(timeout=2.0)
+
+    @patch("core.workflow.graph_engine.worker_management.worker_pool.Worker")
+    def test_worker_pool_uses_shorter_timeout(self, mock_worker_cls: MagicMock):
+        """Test that WorkerPool uses 2s timeout instead of 10s."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+        mock_graph = MagicMock(spec=Graph)
+        mock_graph.nodes = {}
+        mock_graph.edges = {}
+        mock_graph.root_node = MagicMock()
+
+        engine = GraphEngine(
+            workflow_id="test_workflow",
+            graph=mock_graph,
+            graph_runtime_state=runtime_state,
+            command_channel=InMemoryChannel(),
+        )
+
+        worker_pool = engine._worker_pool
+        worker_pool.start(initial_count=1)  # Start with one worker
+
+        mock_worker_instance = mock_worker_cls.return_value
+        mock_worker_instance.is_alive.return_value = True
+
+        worker_pool.stop()
+
+        mock_worker_instance.join.assert_called_once_with(timeout=2.0)
+
+
+class TestStopEventResumeBehavior:
+    """Test stop_event behavior during workflow resume."""
+
+    def test_stop_event_cleared_on_resume(self):
+        """Test that stop_event is cleared when resuming a paused workflow."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+        mock_graph = MagicMock(spec=Graph)
+        mock_graph.nodes = {}
+        mock_graph.edges = {}
+        mock_graph.root_node = MagicMock()
+        mock_graph.root_node.id = "start"  # Set proper id
+
+        start_node = StartNode(
+            id="start",
+            config={"id": "start", "data": {"title": "start", "variables": []}},
+            graph_init_params=GraphInitParams(
+                tenant_id="test_tenant",
+                app_id="test_app",
+                workflow_id="test_workflow",
+                graph_config={},
+                user_id="test_user",
+                user_from=UserFrom.ACCOUNT,
+                invoke_from=InvokeFrom.DEBUGGER,
+                call_depth=0,
+            ),
+            graph_runtime_state=runtime_state,
+        )
+        mock_graph.nodes["start"] = start_node
+        mock_graph.get_outgoing_edges = MagicMock(return_value=[])
+        mock_graph.get_incoming_edges = MagicMock(return_value=[])
+
+        engine = GraphEngine(
+            workflow_id="test_workflow",
+            graph=mock_graph,
+            graph_runtime_state=runtime_state,
+            command_channel=InMemoryChannel(),
+        )
+
+        # Simulate a previous execution that set stop_event
+        engine._stop_event.set()
+        assert engine._stop_event.is_set()
+
+        # Run the engine (should clear stop_event in _start_execution)
+        events = list(engine.run())
+
+        # Execution should complete successfully
+        assert any(isinstance(e, GraphRunStartedEvent) for e in events)
+        assert any(isinstance(e, GraphRunSucceededEvent) for e in events)
+
+
+class TestWorkerStopBehavior:
+    """Test Worker behavior with shared stop_event."""
+
+    def test_worker_uses_shared_stop_event(self):
+        """Test that Worker uses shared stop_event from GraphEngine."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+        mock_graph = MagicMock(spec=Graph)
+        mock_graph.nodes = {}
+        mock_graph.edges = {}
+        mock_graph.root_node = MagicMock()
+
+        engine = GraphEngine(
+            workflow_id="test_workflow",
+            graph=mock_graph,
+            graph_runtime_state=runtime_state,
+            command_channel=InMemoryChannel(),
+        )
+
+        # Get the worker pool and check workers
+        worker_pool = engine._worker_pool
+
+        # Start the worker pool to create workers
+        worker_pool.start()
+
+        # Check that at least one worker was created
+        assert len(worker_pool._workers) > 0
+
+        # Verify workers use the shared stop_event
+        for worker in worker_pool._workers:
+            assert worker._stop_event is engine._stop_event
+
+        # Clean up
+        worker_pool.stop()
+
+    def test_worker_stop_is_noop(self):
+        """Test that Worker.stop() is now a no-op."""
+        runtime_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.perf_counter())
+
+        # Create a mock worker
+        from core.workflow.graph_engine.ready_queue import InMemoryReadyQueue
+        from core.workflow.graph_engine.worker import Worker
+
+        ready_queue = InMemoryReadyQueue()
+        event_queue = MagicMock()
+
+        # Create a proper mock graph with real dict
+        mock_graph = Mock(spec=Graph)
+        mock_graph.nodes = {}  # Use real dict
+
+        stop_event = threading.Event()
+
+        worker = Worker(
+            ready_queue=ready_queue,
+            event_queue=event_queue,
+            graph=mock_graph,
+            layers=[],
+            stop_event=stop_event,
+        )
+
+        # Calling stop() should do nothing (no-op)
+        # and should NOT set the stop_event
+        worker.stop()
+        assert not stop_event.is_set()