diff --git a/src/backend/base/langflow/api/v2/workflow.py b/src/backend/base/langflow/api/v2/workflow.py index 88a7f4a144..f61a163312 100644 --- a/src/backend/base/langflow/api/v2/workflow.py +++ b/src/backend/base/langflow/api/v2/workflow.py @@ -932,15 +932,29 @@ async def get_workflow_status( folder_id=getattr(flow, "folder_id", None), ) - # Reconstruct response from vertex_build table - return await reconstruct_workflow_response_from_job_id( - session=session, - flow=flow, - job_id=job_id_str, - user_id=str(current_user.id), - ) + # Reconstruct response from vertex_build table (sync path persists + # those keyed by job_id). Background runs do not write vertex_builds + # keyed by job_id, so reconstruction finds nothing and raises + # ValueError — fall back to the durable Job.result the runner wrote + # so a completed background run reports completed instead of 500ing. + try: + return await reconstruct_workflow_response_from_job_id( + session=session, + flow=flow, + job_id=job_id_str, + user_id=str(current_user.id), + ) + except ValueError: + return WorkflowExecutionResponse( + flow_id=flow_id_str, + job_id=job_id_str, + status=JobStatus.COMPLETED, + ) if job.status == JobStatus.FAILED: + # Surface the durable error JSON the runner persisted, additively. + # The error column is nullable (a crash before the runner could write + # one leaves it None); the static detail still applies in that case. raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail={ @@ -948,6 +962,7 @@ async def get_workflow_status( "code": "JOB_FAILED", "message": f"Job {job_id_str} has failed execution.", "job_id": job_id_str, + "error_detail": job.error, }, ) diff --git a/src/backend/tests/unit/api/v2/test_workflow_facade.py b/src/backend/tests/unit/api/v2/test_workflow_facade.py index 90bbccd15e..5344c0813f 100644 --- a/src/backend/tests/unit/api/v2/test_workflow_facade.py +++ b/src/backend/tests/unit/api/v2/test_workflow_facade.py @@ -15,8 +15,9 @@ contract so the cutover stays backward compatible: Real client + real flows, no mocking of our own code. """ +import asyncio import json -from uuid import uuid4 +from uuid import UUID, uuid4 import pytest from httpx import AsyncClient @@ -24,6 +25,24 @@ from langflow.services.database.models.flow.model import Flow from lfx.services.deps import session_scope +async def _wait_terminal(job_id: str) -> "object": + """Poll the durable Job row until it reaches a terminal state. + + Condition-based (not a fixed sleep): returns the final status enum. + """ + from langflow.services.database.models.jobs.model import Job, JobStatus + + terminal = {JobStatus.COMPLETED, JobStatus.FAILED, JobStatus.CANCELLED, JobStatus.TIMED_OUT} + for _ in range(200): + async with session_scope() as session: + row = await session.get(Job, UUID(job_id)) + if row is not None and row.status in terminal: + return row.status + await asyncio.sleep(0.05) + pytest.fail(f"job {job_id} did not reach a terminal state") + return None + + def _body(flow_id, *, message: str = "hello", mode: str = "background", protocol: str = "langflow") -> dict: return { "flow_id": str(flow_id), @@ -84,3 +103,93 @@ class TestBackgroundSubmitContract: assert result["links"]["stop"] == "/api/v2/workflows/stop" # New additive link: re-attach events URL must point at the job. assert result["links"]["events"] == f"/api/v2/workflows/{result['job_id']}/events" + + +class TestStatusDurableResultError: + """GET status reads durable result/error written by the runner, additively.""" + + async def test_completed_background_run_status_returns_completed( + self, + client: AsyncClient, + created_api_key, + chatbot_flow, + ): + """A finished background run reports completed via GET status (not a 500). + + The background path does not persist vertex_build rows keyed by job_id, + so the status endpoint must fall back to the durable Job.result rather + than 500 on an empty vertex-build reconstruction. + """ + headers = {"x-api-key": created_api_key.api_key} + start = await client.post( + "api/v2/workflows", + json=_body(chatbot_flow, mode="background"), + headers=headers, + ) + assert start.status_code == 200 + job_id = start.json()["job_id"] + + from langflow.services.database.models.jobs.model import JobStatus + + assert await _wait_terminal(job_id) == JobStatus.COMPLETED + + status_resp = await client.get(f"api/v2/workflows?job_id={job_id}", headers=headers) + + assert status_resp.status_code == 200, status_resp.text + result = status_resp.json() + assert result["status"] == "completed" + assert result["flow_id"] == str(chatbot_flow) + + async def test_failed_background_run_status_carries_durable_error( + self, + client: AsyncClient, + created_api_key, + ): + """A failed run surfaces the durable error JSON additively in the detail. + + The top-level ``error`` string stays unchanged; the stored error blob the + runner persisted rides under ``error_detail`` so the wire contract is + additive-only. + """ + headers = {"x-api-key": created_api_key.api_key} + # A flow whose data is corrupt fails during the background run; the runner + # persists the durable error which status echoes additively. + bad_flow = uuid4() + async with session_scope() as session: + flow = Flow( + id=bad_flow, + name="Facade Bad Flow", + description="Corrupt flow that fails at build time", + data={"nodes": [{"id": "broken"}], "edges": []}, + user_id=created_api_key.user_id, + ) + session.add(flow) + await session.flush() + + try: + start = await client.post( + "api/v2/workflows", + json=_body(bad_flow, mode="background"), + headers=headers, + ) + assert start.status_code == 200 + job_id = start.json()["job_id"] + + from langflow.services.database.models.jobs.model import JobStatus + + assert await _wait_terminal(job_id) == JobStatus.FAILED + + status_resp = await client.get(f"api/v2/workflows?job_id={job_id}", headers=headers) + + assert status_resp.status_code == 500 + detail = status_resp.json()["detail"] + assert detail["code"] == "JOB_FAILED" + assert detail["job_id"] == job_id + # Unchanged top-level error string + additive durable error blob. + assert detail["error"] == "Job failed" + assert detail["error_detail"] is not None + finally: + async with session_scope() as session: + flow = await session.get(Flow, bad_flow) + if flow: + await session.delete(flow)