fix: Generate job response from vertex builds history by job id. (#11457)

* feat: reconstruct workflow execution response from vertex_build by job_id

* [autofix.ci] apply automated fixes

* fix: use correct attribute name 'id' instead of 'vertex_id' in VertexBuildTable

* Updated the GET endpoint to return WorkflowExecutionResponse

---------

Co-authored-by: Janardan S Kavia <[email protected]>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

Author: 3 people

src/backend/base/langflow/api/v2/workflow.py

@@ -39,7 +39,7 @@
     WorkflowStopRequest,
     WorkflowStopResponse,
 )
-from lfx.services.deps import get_settings_service
+from lfx.services.deps import get_settings_service, injectable_session_scope_readonly
 from pydantic_core import ValidationError as PydanticValidationError
 from sqlalchemy.exc import OperationalError
 
@@ -486,9 +486,10 @@ async def execute_workflow_background(
     description="Get status of workflow job by job ID",
 )
 async def get_workflow_status(
-    api_key_user: Annotated[UserRead, Depends(api_key_security)],  # noqa: ARG001
+    api_key_user: Annotated[UserRead, Depends(api_key_security)],
     job_id: Annotated[UUID | None, Query(description="Job ID to query")] = None,
-) -> list[WorkflowJobResponse] | WorkflowJobResponse:
+    session: Annotated[object, Depends(injectable_session_scope_readonly)] = None,
+) -> WorkflowExecutionResponse | WorkflowJobResponse:
     """Get workflow job status and results.
 
     This endpoint allows clients to query job status either by:
@@ -500,6 +501,7 @@ async def get_workflow_status(
     Args:
         api_key_user: Authenticated user from API key
         job_id: Optional job ID to query specific job
+        session: Database session for querying vertex builds
         page: Page number for pagination (default: 1)
         page_size: Number of results per page (default: 10, max: 100)
 
@@ -538,10 +540,32 @@ async def get_workflow_status(
                     },
                 )
             job = await job_service.get_job_by_job_id(job_id=job_id)
-            return WorkflowJobResponse(
-                job_id=str(job.job_id),
+
+            # If job is completed, reconstruct full workflow response from vertex_builds
+            if job.status == JobStatus.COMPLETED:
+                from langflow.api.v2.workflow_reconstruction import reconstruct_workflow_response_from_job_id
+
+                # Get the flow
+                flow = await get_flow_by_id_or_endpoint_name(str(job.flow_id), api_key_user.id)
+
+                # Reconstruct response from vertex_build table
+                return await reconstruct_workflow_response_from_job_id(
+                    session=session,
+                    flow=flow,
+                    job_id=str(job.job_id),
+                    user_id=str(api_key_user.id),
+                )
+
+            # If not completed, return WorkflowExecutionResponse with empty outputs
+            # This ensures consistent response type (always WorkflowExecutionResponse) for SDK generation
+            return WorkflowExecutionResponse(
                 flow_id=str(job.flow_id),
+                job_id=str(job.job_id),
                 status=job.status,
+                outputs={},
+                errors=[],
+                inputs={},
+                metadata={},
             )
 
     except HTTPException:

src/backend/base/langflow/api/v2/workflow_reconstruction.py

@@ -0,0 +1,80 @@
+"""Workflow response reconstruction from vertex_build table.
+
+This module reconstructs WorkflowExecutionResponse from vertex_build table data by job_id,
+enabling retrieval of past execution results without re-running workflows.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from lfx.graph.graph.base import Graph
+from lfx.graph.schema import ResultData, RunOutputs
+from lfx.schema.workflow import WorkflowExecutionRequest
+
+from langflow.api.v1.schemas import RunResponse
+from langflow.api.v2.converters import run_response_to_workflow_response
+from langflow.services.database.models.vertex_builds.crud import get_vertex_builds_by_job_id
+
+if TYPE_CHECKING:
+    from sqlmodel.ext.asyncio.session import AsyncSession
+
+    from langflow.services.database.models.flow.model import FlowRead
+
+
+async def reconstruct_workflow_response_from_job_id(
+    session: AsyncSession,
+    flow: FlowRead,
+    job_id: str,
+    user_id: str,
+):
+    """Reconstruct WorkflowExecutionResponse from vertex_builds by job_id.
+
+    Args:
+        session: Database session (readonly for performance)
+        flow: Flow model from database
+        job_id: Job ID to query vertex builds
+        user_id: User ID for graph construction
+
+    Returns:
+        WorkflowExecutionResponse reconstructed from vertex_build data
+
+    Raises:
+        ValueError: If flow has no data or no vertex builds found for job_id
+    """
+    # Validate flow data
+    if not flow.data:
+        msg = f"Flow {flow.id} has no data"
+        raise ValueError(msg)
+
+    # Query vertex_builds by job_id
+    vertex_builds = await get_vertex_builds_by_job_id(session, job_id)
+    if not vertex_builds:
+        msg = f"No vertex builds found for job_id {job_id}"
+        raise ValueError(msg)
+
+    # Build graph to identify terminal nodes
+    flow_id_str = str(flow.id)
+    graph = Graph.from_payload(flow.data, flow_id=flow_id_str, user_id=user_id, flow_name=flow.name)
+    terminal_node_ids = graph.get_terminal_nodes()
+
+    # Filter to terminal vertices with data
+    terminal_vertex_builds = [vb for vb in vertex_builds if vb.id in terminal_node_ids and vb.data]
+    if not terminal_vertex_builds:
+        msg = f"No terminal vertex builds found for job_id {job_id}"
+        raise ValueError(msg)
+
+    # Convert vertex_build data to RunOutputs format
+    run_outputs_list = [RunOutputs(inputs={}, outputs=[ResultData(**vb.data)]) for vb in terminal_vertex_builds]
+
+    # Create RunResponse and convert to WorkflowExecutionResponse
+    run_response = RunResponse(outputs=run_outputs_list, session_id=None)
+    workflow_request = WorkflowExecutionRequest(flow_id=flow_id_str, inputs={})
+
+    return run_response_to_workflow_response(
+        run_response=run_response,
+        flow_id=flow_id_str,
+        job_id=job_id,
+        workflow_request=workflow_request,
+        graph=graph,
+    )

src/backend/base/langflow/services/database/models/vertex_builds/crud.py

@@ -145,16 +145,19 @@ async def delete_vertex_builds_by_flow_id(db: AsyncSession, flow_id: UUID) -> No
     await db.exec(stmt)
 
 
-async def get_vertex_builds_by_job_id(db: AsyncSession, job_id: str) -> list[VertexBuildTable]:
+async def get_vertex_builds_by_job_id(db: AsyncSession, job_id: str | UUID) -> list[VertexBuildTable]:
     """Get all vertex builds associated with a specific job ID.
 
     Args:
         db (AsyncSession): The database session for executing queries.
-        job_id (str): The unique identifier of the job to get builds for.
+        job_id (str | UUID): The unique identifier of the job to get builds for.
 
     Returns:
         list[VertexBuildTable]: List of vertex builds, ordered chronologically by timestamp.
     """
+    if isinstance(job_id, str):
+        job_id = UUID(job_id)
+
     stmt = select(VertexBuildTable).where(VertexBuildTable.job_id == job_id).order_by(col(VertexBuildTable.timestamp))
 
     builds = await db.exec(stmt)

src/backend/tests/unit/api/v2/test_workflow_reconstruction.py

@@ -0,0 +1,154 @@
+"""Unit tests for workflow reconstruction from vertex_build table.
+
+Test Coverage:
+    - Successful reconstruction with terminal nodes
+    - Reconstruction with no vertex builds found (error case)
+    - Reconstruction with flow having no data (error case)
+    - Reconstruction filtering to terminal nodes only
+"""
+
+from datetime import datetime, timezone
+from unittest.mock import MagicMock, patch
+from uuid import uuid4
+
+import pytest
+from langflow.api.v2.workflow_reconstruction import reconstruct_workflow_response_from_job_id
+from langflow.services.database.models.vertex_builds.model import VertexBuildTable
+
+
+class TestWorkflowReconstruction:
+    """Unit tests for workflow reconstruction logic."""
+
+    async def test_reconstruct_success_with_terminal_nodes(self):
+        """Test successful reconstruction filters to terminal nodes and returns response."""
+        flow_id = uuid4()
+        job_id = uuid4()
+        user_id = uuid4()
+
+        # Mock flow
+        mock_flow = MagicMock()
+        mock_flow.id = flow_id
+        mock_flow.data = {"nodes": [{"id": "node1"}, {"id": "node2"}], "edges": []}
+
+        # Mock vertex_builds
+        mock_vb1 = MagicMock(spec=VertexBuildTable)
+        mock_vb1.id = "node1"
+        mock_vb1.data = {"outputs": {"result": "output1"}}
+        mock_vb1.artifacts = {}
+        mock_vb1.timestamp = datetime.now(timezone.utc)
+
+        mock_vb2 = MagicMock(spec=VertexBuildTable)
+        mock_vb2.id = "node2"
+        mock_vb2.data = {"outputs": {"result": "output2"}}
+        mock_vb2.artifacts = {}
+        mock_vb2.timestamp = datetime.now(timezone.utc)
+
+        mock_session = MagicMock()
+
+        with (
+            patch("langflow.api.v2.workflow_reconstruction.get_vertex_builds_by_job_id") as mock_get_vb,
+            patch("langflow.api.v2.workflow_reconstruction.Graph") as mock_graph_class,
+            patch("langflow.api.v2.workflow_reconstruction.run_response_to_workflow_response") as mock_converter,
+        ):
+            mock_get_vb.return_value = [mock_vb1, mock_vb2]
+
+            mock_graph = MagicMock()
+            mock_graph.get_terminal_nodes.return_value = ["node1", "node2"]
+            mock_graph_class.from_payload.return_value = mock_graph
+
+            mock_response = MagicMock()
+            mock_response.flow_id = str(flow_id)
+            mock_response.job_id = str(job_id)
+            mock_converter.return_value = mock_response
+
+            result = await reconstruct_workflow_response_from_job_id(
+                session=mock_session,
+                flow=mock_flow,
+                job_id=str(job_id),
+                user_id=user_id,
+            )
+
+            assert result.flow_id == str(flow_id)
+            assert result.job_id == str(job_id)
+            mock_get_vb.assert_called_once_with(mock_session, str(job_id))
+            mock_graph.get_terminal_nodes.assert_called_once()
+
+    async def test_reconstruct_fails_when_no_vertex_builds(self):
+        """Test reconstruction raises ValueError when no vertex_builds found."""
+        mock_flow = MagicMock()
+        mock_flow.data = {"nodes": [{"id": "node1"}], "edges": []}
+        mock_session = MagicMock()
+
+        with patch("langflow.api.v2.workflow_reconstruction.get_vertex_builds_by_job_id") as mock_get_vb:
+            mock_get_vb.return_value = []
+
+            with pytest.raises(ValueError, match="No vertex builds found"):
+                await reconstruct_workflow_response_from_job_id(
+                    session=mock_session,
+                    flow=mock_flow,
+                    job_id=str(uuid4()),
+                    user_id=uuid4(),
+                )
+
+    async def test_reconstruct_fails_when_flow_has_no_data(self):
+        """Test reconstruction raises ValueError when flow has no data."""
+        mock_flow = MagicMock()
+        mock_flow.data = None
+        mock_session = MagicMock()
+
+        with pytest.raises(ValueError, match="has no data"):
+            await reconstruct_workflow_response_from_job_id(
+                session=mock_session,
+                flow=mock_flow,
+                job_id=str(uuid4()),
+                user_id=uuid4(),
+            )
+
+    async def test_reconstruct_filters_to_terminal_nodes_only(self):
+        """Test reconstruction only includes terminal node outputs, not intermediate nodes."""
+        flow_id = uuid4()
+        job_id = uuid4()
+        user_id = uuid4()
+
+        mock_flow = MagicMock()
+        mock_flow.id = flow_id
+        mock_flow.data = {"nodes": [{"id": "node1"}, {"id": "node2"}, {"id": "node3"}], "edges": []}
+
+        # Create vertex_builds for all 3 nodes
+        mock_vertex_builds = []
+        for node_id in ["node1", "node2", "node3"]:
+            mock_vb = MagicMock(spec=VertexBuildTable)
+            mock_vb.id = node_id
+            mock_vb.data = {"outputs": {"result": f"output_{node_id}"}}
+            mock_vb.artifacts = {}
+            mock_vb.timestamp = datetime.now(timezone.utc)
+            mock_vertex_builds.append(mock_vb)
+
+        mock_session = MagicMock()
+
+        with (
+            patch("langflow.api.v2.workflow_reconstruction.get_vertex_builds_by_job_id") as mock_get_vb,
+            patch("langflow.api.v2.workflow_reconstruction.Graph") as mock_graph_class,
+            patch("langflow.api.v2.workflow_reconstruction.run_response_to_workflow_response") as mock_converter,
+        ):
+            mock_get_vb.return_value = mock_vertex_builds
+
+            # Only node1 and node3 are terminal nodes (node2 is intermediate)
+            mock_graph = MagicMock()
+            mock_graph.get_terminal_nodes.return_value = ["node1", "node3"]
+            mock_graph_class.from_payload.return_value = mock_graph
+
+            mock_response = MagicMock()
+            mock_converter.return_value = mock_response
+
+            result = await reconstruct_workflow_response_from_job_id(
+                session=mock_session,
+                flow=mock_flow,
+                job_id=str(job_id),
+                user_id=user_id,
+            )
+
+            assert result is not None
+            mock_converter.assert_called_once()
+            # Verify filtering happened by checking terminal nodes were retrieved
+            mock_graph.get_terminal_nodes.assert_called_once()