fix: use thread local isolation the context (#31410)

api/core/workflow/context/execution_context.py

@@ -3,6 +3,7 @@ Execution Context - Abstracted context management for workflow execution.
 """
 
 import contextvars
+import threading
 from abc import ABC, abstractmethod
 from collections.abc import Callable, Generator
 from contextlib import AbstractContextManager, contextmanager
@@ -88,6 +89,7 @@ class ExecutionContext:
         self._app_context = app_context
         self._context_vars = context_vars
         self._user = user
+        self._local = threading.local()
 
     @property
     def app_context(self) -> AppContext | None:
@@ -125,14 +127,16 @@ class ExecutionContext:
 
     def __enter__(self) -> "ExecutionContext":
         """Enter the execution context."""
-        self._cm = self.enter()
-        self._cm.__enter__()
+        cm = self.enter()
+        self._local.cm = cm
+        cm.__enter__()
         return self
 
     def __exit__(self, *args: Any) -> None:
         """Exit the execution context."""
-        if hasattr(self, "_cm"):
-            self._cm.__exit__(*args)
+        cm = getattr(self._local, "cm", None)
+        if cm is not None:
+            cm.__exit__(*args)
 
 
 class NullAppContext(AppContext):

api/core/workflow/graph_engine/worker.py

@@ -11,7 +11,6 @@ import time
 from collections.abc import Sequence
 from datetime import datetime
 from typing import TYPE_CHECKING, final
-from uuid import uuid4
 
 from typing_extensions import override
 
@@ -113,7 +112,7 @@ class Worker(threading.Thread):
                 self._ready_queue.task_done()
             except Exception as e:
                 error_event = NodeRunFailedEvent(
-                    id=str(uuid4()),
+                    id=node.execution_id,
                     node_id=node.id,
                     node_type=node.node_type,
                     in_iteration_id=None,