diff --git a/src/backend/base/langflow/api/v2/converters.py b/src/backend/base/langflow/api/v2/converters.py index d6a334f7c8..d51a86a525 100644 --- a/src/backend/base/langflow/api/v2/converters.py +++ b/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, diff --git a/src/backend/base/langflow/api/v2/workflow.py b/src/backend/base/langflow/api/v2/workflow.py index 1ec7be2162..17d0158b54 100644 --- a/src/backend/base/langflow/api/v2/workflow.py +++ b/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: diff --git a/src/backend/base/langflow/services/background_execution/runner.py b/src/backend/base/langflow/services/background_execution/runner.py index 3fbf0775b6..7ffbc7b7cc 100644 --- a/src/backend/base/langflow/services/background_execution/runner.py +++ b/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) diff --git a/src/backend/tests/unit/api/v2/test_workflow_background.py b/src/backend/tests/unit/api/v2/test_workflow_background.py index b3020aa015..96b273762e 100644 --- a/src/backend/tests/unit/api/v2/test_workflow_background.py +++ b/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.