fix(api/v2): return background run output from completed status

A completed background run's GET status returned a bare COMPLETED with an
empty `outputs` and a null `output`. The COMPLETED branch reconstructs
from `vertex_builds` keyed by job_id, which the durable path does not
write, so reconstruction raised ValueError and fell through to an empty
response; the result was only retrievable via a /events re-attach.

The runner now captures the terminal `output` events (the langflow
adapter's normalized ComponentOutput payloads) into `Job.result`, and the
status COMPLETED branch rebuilds the `outputs` map and resolved `output`
from them via `workflow_response_from_output_events`, matching the sync
response. agui-protocol runs emit no `output` events, so their status
stays result-less (the result remains on the /events log).

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

@@ -464,6 +464,45 @@ def _resolve_output(outputs: dict[str, ComponentOutput], selected_ids: list[str]
     return WorkflowOutput(reason=reason)
 
 
+def workflow_response_from_output_events(
+    output_events: list[dict[str, Any]],
+    *,
+    flow_id: str,
+    job_id: str,
+) -> WorkflowExecutionResponse:
+    """Rebuild a completed-run response from durable ``output`` event payloads.
+
+    Background runs do not persist ``vertex_builds`` keyed by ``job_id``, so the
+    vertex-build reconstruction path finds nothing. The durable runner instead
+    captures each terminal ``output`` event the langflow adapter emits (an
+    ``OutputEvent``: a ``ComponentOutput`` plus its ``component_id``) into
+    ``Job.result``. Re-keying those by component id reproduces the same
+    ``outputs`` map and resolved ``output`` that sync returns, so a completed
+    background run's GET status carries its result without a /events re-attach.
+    """
+    outputs: dict[str, ComponentOutput] = {}
+    for item in output_events:
+        if not isinstance(item, dict):
+            continue
+        component_id = item.get("component_id")
+        if not component_id:
+            continue
+        fields = {key: value for key, value in item.items() if key != "component_id"}
+        try:
+            outputs[component_id] = ComponentOutput(**fields)
+        except ValueError:
+            # A malformed stored payload should not 500 the status read; skip it
+            # and report whatever outputs did rebuild cleanly.
+            continue
+    return WorkflowExecutionResponse(
+        flow_id=flow_id,
+        job_id=job_id,
+        status=JobStatus.COMPLETED,
+        output=_resolve_output(outputs),
+        outputs=outputs,
+    )
+
+
 def run_response_to_workflow_response(
     run_response: RunResponse,
     flow_id: str,

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

@@ -69,6 +69,7 @@ from langflow.api.v2.converters import (
     create_error_response,
     parse_workflow_run_request,
     run_response_to_workflow_response,
+    workflow_response_from_output_events,
 )
 from langflow.api.v2.workflow_reconstruction import reconstruct_workflow_response_from_job_id
 from langflow.exceptions.api import (
@@ -978,10 +979,15 @@ async def get_workflow_status(
                     user_id=str(current_user.id),
                 )
             except ValueError:
-                return WorkflowExecutionResponse(
+                # Rebuild the result from the ``output`` events the runner
+                # captured into ``Job.result`` (langflow-protocol runs). Falls
+                # back to a bare COMPLETED when none were captured (e.g. an
+                # agui-protocol run, where the result lives only on /events).
+                result = job.result if isinstance(job.result, dict) else {}
+                return workflow_response_from_output_events(
+                    result.get("outputs") or [],
                     flow_id=flow_id_str,
                     job_id=job_id_str,
-                    status=JobStatus.COMPLETED,
                 )
 
         if job.status == JobStatus.FAILED:

src/backend/base/langflow/services/background_execution/runner.py

@@ -191,6 +191,14 @@ class JobRunner:
         """The wrapped coroutine: stream frames, persist, publish, finalize result/error."""
         last_durable_seq = 0
         errored_payload: dict[str, Any] | None = None
+        # Terminal outputs the run produced, captured so GET status can return
+        # the result without forcing a /events re-attach. The langflow adapter
+        # normalizes each terminal output into a durable ``output`` event whose
+        # ``data`` is the same ``OutputEvent`` (a ``ComponentOutput`` plus its
+        # component id) that sync returns in ``outputs[id]``. The agui adapter
+        # does not emit these, so agui-protocol runs leave this empty and their
+        # status stays result-less (the result is still on the /events log).
+        output_events: list[dict[str, Any]] = []
         async for frame_bytes, event_type in self._frame_source(**source_kwargs):
             if self._adapter.is_durable(event_type):
                 # Vertex/milestone-boundary cooperative cancel: a STOP written to
@@ -206,6 +214,10 @@ class JobRunner:
                 await self._bus.publish(str(job_id), LiveFrame(seq=seq, data=self._restamp_id(frame_bytes, seq)))
                 if event_type == self._adapter.terminal_error_type:
                     errored_payload = payload
+                elif event_type == "output":
+                    output_data = payload.get("data")
+                    if isinstance(output_data, dict):
+                        output_events.append(output_data)
             else:
                 await self._bus.publish(
                     str(job_id),
@@ -225,7 +237,7 @@ class JobRunner:
             # Surface as a failure so execute_with_status writes FAILED.
             msg = "Background job emitted a terminal error event"
             raise RuntimeError(msg)
-        await self._jobs.set_result(job_id, {"status": "completed"})
+        await self._jobs.set_result(job_id, {"status": "completed", "outputs": output_events})
 
     async def _stop_requested(self, job_id: UUID) -> bool:
         signals = await self._jobs.unconsumed_signals(job_id)

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

@@ -98,6 +98,40 @@ async def test_background_reaches_terminal_status(client, created_api_key, bg_fl
     assert final == JobStatus.COMPLETED, f"job did not complete: last={final}"
 
 
+async def test_background_status_returns_output(client, created_api_key, bg_flow):
+    """A completed background run's GET status carries its terminal output.
+
+    Background runs do not persist ``vertex_builds`` keyed by job_id, so the
+    vertex-build reconstruction finds nothing. The runner instead captures the
+    terminal ``output`` events into ``Job.result`` and the COMPLETED branch
+    rebuilds the ``outputs`` map from them. Regression: the status previously
+    returned a bare COMPLETED with an empty ``outputs`` and a null ``output``.
+    """
+    from uuid import UUID
+
+    from langflow.services.database.models.jobs.model import Job, JobStatus
+
+    submit = await client.post("api/v2/workflows", json=_body(bg_flow), headers=_headers(created_api_key))
+    job_id = submit.json()["job_id"]
+
+    row = None
+    for _ in range(150):
+        async with session_scope() as session:
+            row = await session.get(Job, UUID(job_id))
+        if row is not None and row.status == JobStatus.COMPLETED:
+            break
+        await asyncio.sleep(0.1)
+    assert row is not None, "job row missing"
+    assert row.status == JobStatus.COMPLETED, "job never completed"
+
+    status = await client.get("api/v2/workflows", params={"job_id": job_id}, headers=_headers(created_api_key))
+    assert status.status_code == 200, status.text
+    body = status.json()
+    assert body["status"] == "completed"
+    # The terminal outputs are present (an empty dict before the fix).
+    assert body["outputs"], f"completed background status carried no outputs: {body}"
+
+
 async def test_stop_does_not_overwrite_completed_job(client, created_api_key, bg_flow):
     """A late ``/stop`` on an already-COMPLETED job must NOT flip it to CANCELLED.