feat(api): add an API to retrieve the last execution of a given node.

api/controllers/console/app/workflow.py

@@ -731,6 +731,27 @@ class WorkflowByIdApi(Resource):
         return None, 204
 
 
+class DraftWorkflowNodeLastRunApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    @marshal_with(workflow_run_node_execution_fields)
+    def get(self, app_model: App, node_id: str):
+        srv = WorkflowService()
+        workflow = srv.get_draft_workflow(app_model)
+        if not workflow:
+            raise NotFound("Workflow not found")
+        node_exec = srv.get_node_last_run(
+            app_model=app_model,
+            workflow=workflow,
+            node_id=node_id,
+        )
+        if node_exec is None:
+            raise NotFound("last run not found")
+        return node_exec
+
+
 api.add_resource(
     DraftWorkflowApi,
     "/apps/<uuid:app_id>/workflows/draft",
@@ -795,3 +816,7 @@ api.add_resource(
     WorkflowByIdApi,
     "/apps/<uuid:app_id>/workflows/<string:workflow_id>",
 )
+api.add_resource(
+    DraftWorkflowNodeLastRunApi,
+    "/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/last-run",
+)

api/services/workflow_service.py

@@ -46,6 +46,22 @@ class WorkflowService:
     Workflow Service
     """
 
+    def get_node_last_run(self, app_model: App, workflow: Workflow, node_id: str) -> WorkflowNodeExecution | None:
+        # TODO(QuantumGhost): This query is not fully covered by index.
+        criteria = (
+            WorkflowNodeExecution.tenant_id == app_model.tenant_id,
+            WorkflowNodeExecution.app_id == app_model.id,
+            WorkflowNodeExecution.workflow_id == workflow.id,
+            WorkflowNodeExecution.node_id == node_id,
+        )
+        node_exec = (
+            db.session.query(WorkflowNodeExecution)
+            .filter(*criteria)
+            .order_by(WorkflowNodeExecution.created_at.desc())
+            .first()
+        )
+        return node_exec
+
     def get_draft_workflow(self, app_model: App) -> Optional[Workflow]:
         """
         Get draft workflow