diff --git a/api/core/repositories/sqlalchemy_workflow_node_execution_repository.py b/api/core/repositories/sqlalchemy_workflow_node_execution_repository.py
index b213449be5..8d916a19db 100644
--- a/api/core/repositories/sqlalchemy_workflow_node_execution_repository.py
+++ b/api/core/repositories/sqlalchemy_workflow_node_execution_repository.py
@@ -103,10 +103,7 @@ class SQLAlchemyWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository)
         inputs = db_model.inputs_dict
         process_data = db_model.process_data_dict
         outputs = db_model.outputs_dict
-        if db_model.execution_metadata_dict:
-            metadata = {NodeRunMetadataKey(k): v for k, v in db_model.execution_metadata_dict.items()}
-        else:
-            metadata = {}
+        metadata = {NodeRunMetadataKey(k): v for k, v in db_model.execution_metadata_dict.items()}
 
         # Convert status to domain enum
         status = NodeExecutionStatus(db_model.status)
diff --git a/api/models/workflow.py b/api/models/workflow.py
index 8751fd1aa8..fc6de8692c 100644
--- a/api/models/workflow.py
+++ b/api/models/workflow.py
@@ -666,8 +666,11 @@ class WorkflowNodeExecution(Base):
         return json.loads(self.process_data) if self.process_data else None
 
     @property
-    def execution_metadata_dict(self) -> dict[str, Any] | None:
-        return json.loads(self.execution_metadata) if self.execution_metadata else None
+    def execution_metadata_dict(self) -> dict[str, Any]:
+        # When the metadata is unset, we return an empty dictionary instead of `None`.
+        # This approach streamlines the logic for the caller, making it easier to handle
+        # cases where metadata is absent.
+        return json.loads(self.execution_metadata) if self.execution_metadata else {}
 
     @property
     def extras(self):
diff --git a/api/tests/unit_tests/models/test_workflow.py b/api/tests/unit_tests/models/test_workflow.py
index fe56f18f1b..70ce224eb6 100644
--- a/api/tests/unit_tests/models/test_workflow.py
+++ b/api/tests/unit_tests/models/test_workflow.py
@@ -1,10 +1,11 @@
+import json
 from unittest import mock
 from uuid import uuid4
 
 import contexts
 from constants import HIDDEN_VALUE
 from core.variables import FloatVariable, IntegerVariable, SecretVariable, StringVariable
-from models.workflow import Workflow
+from models.workflow import Workflow, WorkflowNodeExecution
 
 
 def test_environment_variables():
@@ -137,3 +138,14 @@ def test_to_dict():
         workflow_dict = workflow.to_dict(include_secret=True)
         assert workflow_dict["environment_variables"][0]["value"] == "secret"
         assert workflow_dict["environment_variables"][1]["value"] == "text"
+
+
+class TestWorkflowNodeExecution:
+    def test_execution_metadata_dict(self):
+        node_exec = WorkflowNodeExecution()
+        node_exec.execution_metadata = None
+        assert node_exec.execution_metadata_dict == {}
+
+        original = {"a": 1, "b": ["2"]}
+        node_exec.execution_metadata = json.dumps(original)
+        assert node_exec.execution_metadata_dict == original