feat(workflow-api): persist sync run outputs to Job.result

A sync run already creates a Job row (to support HITL suspend + run_id-keyed
vertex builds) and reaches COMPLETED, but never wrote Job.result — so a later
GET status on its job_id could not return the outputs the request returned
inline. Persist the terminal outputs in the same list-of-OutputEvent shape the
background runner writes, so the status read is protocol-uniform across sync and
background. Best-effort: the caller already holds the response inline, so a
result-cache write failure is logged and never fails the run.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: 2 people

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

@@ -452,6 +452,25 @@ async def execute_sync_workflow_with_timeout(
         raise WorkflowTimeoutError from e
 
 
+async def _persist_sync_result(job_service, job_id: UUID, workflow_response, flow_id) -> None:
+    """Best-effort cache of a completed sync run's outputs into ``Job.result``.
+
+    Stores the same ``{component_id, ...ComponentOutput}`` list shape the background
+    runner writes, so a later GET status on this sync job_id rebuilds an identical
+    response via ``workflow_response_from_output_events``. The caller already holds
+    the response inline, so a persistence failure must never fail the run — it is
+    logged and swallowed.
+    """
+    try:
+        output_events = [
+            {"component_id": component_id, **output.model_dump(mode="json")}
+            for component_id, output in (workflow_response.outputs or {}).items()
+        ]
+        await job_service.set_result(job_id, {"status": "completed", "outputs": output_events})
+    except (RuntimeError, ValueError, OSError):
+        await logger.awarning("Sync result persistence failed for flow %s", flow_id, exc_info=True)
+
+
 async def execute_sync_workflow(
     parsed: ParsedWorkflowRun,
     flow: FlowRead,
@@ -576,7 +595,7 @@ async def execute_sync_workflow(
         # Build RunResponse
         run_response = RunResponse(outputs=task_result, session_id=execution_session_id)
         # Convert to WorkflowExecutionResponse
-        return run_response_to_workflow_response(
+        workflow_response = run_response_to_workflow_response(
             run_response=run_response,
             flow_id=parsed.flow_id,
             job_id=str(job_id),
@@ -585,6 +604,11 @@ async def execute_sync_workflow(
             effective_globals=request_variables,
             selected_ids=parsed.output_ids,
         )
+        # Persist the terminal outputs to Job.result so a later GET status on this
+        # sync job_id returns the same outputs the background path stores (same
+        # list-of-OutputEvent shape, so the status read is protocol-uniform).
+        await _persist_sync_result(job_service, job_id, workflow_response, flow.id)
+        return workflow_response  # noqa: TRY300 — keep response-building under the broad except below
 
     except GraphPausedException as exc:
         # HITL: a pausing node suspended the run for human input. The checkpoint is already

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

@@ -341,3 +341,38 @@ async def test_finalize_job_status_still_writes_terminal_for_running_job():
         await wb._finalize_job_status(uuid4(), JobStatus.COMPLETED)
 
     fake_service.update_job_status.assert_awaited_once()
+
+
+async def test_sync_run_persists_job_result(client, created_api_key, bg_flow):
+    """A sync run persists its outputs to Job.result so a later GET status returns them.
+
+    Sync already creates a Job row (to support HITL suspend + run_id-keyed builds);
+    this asserts the completed run now also writes Job.result in the same
+    list-of-OutputEvent shape the background path stores, so the status read is
+    protocol-uniform across sync and background.
+    """
+    from uuid import UUID
+
+    from langflow.services.database.models.jobs.model import Job
+
+    body = {"flow_id": bg_flow, "mode": "sync", "input_value": "hi"}
+    resp = await client.post("api/v2/workflows", json=body, headers=_headers(created_api_key))
+    assert resp.status_code == 200, resp.text
+    data = resp.json()
+    assert data["status"] == "completed", data
+    assert data["outputs"], f"sync response carried no outputs: {data}"
+    job_id = data["job_id"]
+
+    # Job.result persisted (same {status, outputs:[...]} blob the runner writes).
+    async with session_scope() as session:
+        row = await session.get(Job, UUID(job_id))
+    assert row is not None
+    assert isinstance(row.result, dict), f"Job.result not a dict: {row.result!r}"
+    assert row.result.get("outputs"), f"Job.result has no outputs: {row.result}"
+
+    # GET status reconstructs the same outputs from Job.result.
+    status = await client.get("api/v2/workflows", params={"job_id": job_id}, headers=_headers(created_api_key))
+    assert status.status_code == 200, status.text
+    sbody = status.json()
+    assert sbody["status"] == "completed"
+    assert sbody["outputs"], f"GET status carried no outputs for sync job: {sbody}"