feat: Remove use of vertex builds table in workflow API background mode (langflow-ai#14353)

* feat(workflow-api): make Job.result the default GET-status source with session_id

Flip the v2 workflow GET-status construction so the durable Job.result blob
is the primary output source and vertex_build reconstruction is the fallback.
Echo session_id (from job_metadata request, falling back to flow id) on the
response so completed background runs return the same shape as sync.

Populate Job.result for AG-UI runs via protocol-neutral off-wire capture: a
synthesized WORKFLOW_OUTPUT_CAPTURE_EVENT frame built from the raw end_vertex
before adapter.translate, captured in-memory only (never appended or published,
so it cannot leak to /events). This makes AG-UI runs carry full outputs even
though the wire protocol emits no output event.

Add LANGFLOW_JOB_EVENTS_STORAGE_ENABLED to gate job_events persistence. When
off, the runner uses a local seq fallback so live subscribers still work while
per-milestone DB writes and reattach/replay are skipped; Job.result and
GET-status output are unaffected.

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

fix(workflow-api): preserve pre-pause HITL outputs; drop job-events storage flag

A background run that produced a terminal output on one branch, then paused
for human input on another, lost that output: the resumed pass starts a fresh
capture list and the output loop skips re-emitting checkpoint-restored
vertices, so finalize only had the resumed pass's captures.

Fix: PauseRequested carries the pre-pause output_events out of _drive; _suspend
stashes them (deduped) in job_metadata.pre_pause_outputs; a resumed _drive
pre-seeds its capture list from that stash; set_result dedups by component_id
(first position, latest value) so a re-emitted vertex overwrites its stale
entry and branch order is preserved.

Also remove the job_events_storage_enabled setting — the durable event log is
now always persisted (removes the event-storage-off path that dropped live
HITL requests). Drops the field from EXPECTED_FIELDS and the now-impossible
off-mode test.

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

fix: harden background workflow result handling

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>

Fixed exception handling upon persistence failure but execution success. now persistence issue won't block users from getting the output if the workflow ran successfully.

* fix: satisfy Ruff in workflow background test

---------

Co-authored-by: Janardan S Kavia <janardanskavia@Janardans-MacBook-Pro.local>
Co-authored-by: Eric Hare <ericrhare@gmail.com>

Author: 3 people

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

@@ -458,6 +458,10 @@ async def _source(*, job_id=None, resume=None, **_kwargs):
                 # job_id) and fires the memory-base hook below with that id, so the build pipeline
                 # must not mint its own run_id-keyed WORKFLOW row + hook (it would double both).
                 track_job_status=False,
+                # Emit the off-wire terminal-output capture the runner records into
+                # ``Job.result`` — protocol-neutral, so agui-protocol runs get a
+                # populated GET-status result too (not just langflow).
+                emit_output_capture=True,
             ):
                 if terminal_error_type is not None and event_type == terminal_error_type:
                     errored = True
@@ -644,29 +648,64 @@ async def get_workflow_status(
                 folder_id=getattr(flow, "folder_id", None),
             )
 
-            # 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.
+            # The session the run executed under, resolved exactly as the runner's
+            # frame source did (``parsed.session_id or str(flow.id)``): the submit
+            # request is persisted on ``job_metadata["request"]``, so a completed
+            # background GET echoes the same chat/memory thread sync returns. The
+            # terminal output's ``content`` is a rendered string, so it can't be
+            # searched structurally — the persisted request is the source of truth.
+            persisted_request = (job.job_metadata or {}).get("request") or {}
+            effective_session_id = persisted_request.get("session_id") or flow_id_str
+
+            # Default GET-status path: rebuild from the protocol-neutral terminal
+            # captures the runner stored in ``Job.result``. This needs no
+            # ``vertex_build`` rows, so it works with vertex-build storage off
+            # (headless) and skips graph reconstruction.
+            result = job.result if isinstance(job.result, dict) else {}
+            output_events = result.get("outputs") or []
+            partial_stored_response: WorkflowExecutionResponse | None = None
+            if isinstance(output_events, list) and output_events:
+                try:
+                    return workflow_response_from_output_events(
+                        output_events,
+                        flow_id=flow_id_str,
+                        job_id=job_id_str,
+                        session_id=effective_session_id,
+                        fail_on_rejected=True,
+                    )
+                except ValueError:
+                    # Preserve every decodable capture in case the legacy
+                    # vertex-build fallback is also unavailable.
+                    partial_stored_response = workflow_response_from_output_events(
+                        output_events,
+                        flow_id=flow_id_str,
+                        job_id=job_id_str,
+                        session_id=effective_session_id,
+                    )
+
+            # Fallback: ``Job.result`` carried no outputs, has an invalid shape,
+            # contains rejected/version-skewed entries, or predates capture.
+            # Reconstruct from ``vertex_build`` rows keyed by job_id when present.
             try:
-                return await reconstruct_workflow_response_from_job_id(
+                reconstructed = await reconstruct_workflow_response_from_job_id(
                     session=session,
                     flow=flow,
                     job_id=job_id_str,
                     user_id=str(current_user.id),
                 )
             except ValueError:
-                # 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 {}
+                if partial_stored_response is not None:
+                    return partial_stored_response
                 return workflow_response_from_output_events(
-                    result.get("outputs") or [],
+                    [],
                     flow_id=flow_id_str,
                     job_id=job_id_str,
+                    session_id=effective_session_id,
                 )
+            else:
+                if reconstructed.session_id is None:
+                    reconstructed = reconstructed.model_copy(update={"session_id": effective_session_id})
+                return reconstructed
 
         if job.status == JobStatus.FAILED:
             # Surface the durable error JSON the runner persisted, additively.

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

@@ -37,6 +37,7 @@
 from lfx.schema.schema import InputValueRequest
 from lfx.schema.workflow import JobStatus, WorkflowExecutionResponse
 from lfx.workflow.adapters import StreamAdapter, StreamEvent
+from lfx.workflow.adapters.langflow import WORKFLOW_OUTPUT_CAPTURE_EVENT, build_terminal_output_event
 from lfx.workflow.converters import ParsedWorkflowRun, create_error_response, run_response_to_workflow_response
 
 from langflow.api.utils import extract_global_variables_from_headers
@@ -185,6 +186,7 @@ async def _stream_event_frames(
     job_id: UUID | None = None,
     resume: dict | None = None,
     track_job_status: bool = True,
+    emit_output_capture: bool = False,
 ) -> AsyncIterator[tuple[bytes, str]]:
     """Run a flow via the v1 build-vertex loop, dispatch its events through ``adapter``.
 
@@ -321,6 +323,26 @@ def _frame(stream_event: StreamEvent, seq: int) -> tuple[bytes, str]:
                     seq,
                 )
                 seq += 1
+            # Off-wire terminal-output capture for ``Job.result`` (background only).
+            # Synthesized from the RAW ``end_vertex`` here — before ``adapter.translate``
+            # — so it is protocol-neutral: the ``agui`` adapter emits no wire ``output``
+            # event, so without this its background GET-status would carry no outputs.
+            # The runner captures this frame in-memory only (never persisted to
+            # ``job_events``, never published), so the wire is unchanged for every
+            # protocol and the capture is independent of durable-event storage.
+            if emit_output_capture and event_type == "end_vertex":
+                output = build_terminal_output_event(event_data)
+                if output is not None:
+                    capture_payload = {"event": "output", "data": output.model_dump(mode="json")}
+                    yield _frame(
+                        StreamEvent(
+                            type=WORKFLOW_OUTPUT_CAPTURE_EVENT,
+                            data_json=json.dumps(capture_payload, default=str),
+                        ),
+                        seq,
+                    )
+                    seq += 1
+
             for event in adapter.translate(event_type, event_data):
                 if terminal_error_type is not None and event.type == terminal_error_type:
                     terminal_error_seen = True
@@ -430,6 +452,33 @@ 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.
+
+    The run has already executed and succeeded upstream; this helper only caches
+    the finished result, so EVERY exception it can raise (serialization or the
+    ``set_result`` DB write — e.g. a SQLite ``OperationalError`` under lock
+    contention) is a persistence failure, never a workflow failure. Catch broadly
+    so such an error cannot escape to the caller's terminal ``except Exception`` and
+    be misreported as a failed run. ``asyncio.CancelledError`` is a ``BaseException``
+    and still propagates, so timeouts/disconnects are unaffected.
+    """
+    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 Exception:  # noqa: BLE001 — best-effort cache; the response is already built inline
+        await logger.awarning("Sync result persistence failed for flow %s", flow_id, exc_info=True)
+
+
 async def execute_sync_workflow(
     parsed: ParsedWorkflowRun,
     flow: FlowRead,
@@ -554,7 +603,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),
@@ -563,6 +612,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/base/langflow/services/background_execution/runner.py

@@ -23,6 +23,7 @@
 from typing import TYPE_CHECKING, Any
 
 from lfx.log.logger import logger
+from lfx.workflow.adapters.langflow import WORKFLOW_OUTPUT_CAPTURE_EVENT
 
 from langflow.services.background_execution.live_bus import LiveFrame
 from langflow.services.database.models.jobs.model import JobStatus, SignalType
@@ -146,15 +147,21 @@ async def _suspend(self, job_id: UUID, exc: PauseRequested) -> None:
         finalizer, or close the live bus — the run is resumable.
         """
         await self._jobs.append_event(job_id, HUMAN_INPUT_REQUIRED_EVENT, exc.payload)
-        await self._jobs.update_job_status(job_id, JobStatus.SUSPENDED)
         metadata: dict[str, Any] = {}
         if exc.request_id is not None:
             metadata["pending_request_id"] = exc.request_id
         if self._input_deadline_s is not None:
             deadline = datetime.now(timezone.utc) + timedelta(seconds=self._input_deadline_s)
             metadata["input_deadline_at"] = deadline.isoformat()
-        if metadata:
-            await self._jobs.update_job_metadata(job_id, metadata)
+        if exc.output_events:
+            # Durable stash of the pre-pause terminal outputs so the resumed pass can
+            # pre-seed its capture list. Deduped so a resume-pause-resume cycle does
+            # not accumulate duplicates (the pre-seed already folds the prior stash in).
+            metadata["pre_pause_outputs"] = self._dedup_outputs(exc.output_events)
+        # Publish SUSPENDED only when the resume guard and pre-pause captures are
+        # visible in the same committed row. Otherwise a racing resume can claim
+        # the job between separate status and metadata transactions.
+        await self._jobs.suspend_job(job_id, metadata)
 
     async def _finalize_terminal_event(self, job_id: UUID) -> None:
         """Backfill the error blob + terminal event for TIMED_OUT / CANCELLED.
@@ -232,20 +239,42 @@ async def _drive(self, *, job_id: UUID, source_kwargs: dict[str, Any]) -> None:
         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]] = []
+        # the result without forcing a /events re-attach. The frame source emits
+        # a protocol-neutral off-wire capture for both langflow and AG-UI. The
+        # public langflow ``output`` event is also accepted as a fallback for
+        # injected/exported FrameSource implementations that do not emit the
+        # private capture; final deduplication keeps the two paths from doubling.
+        #
+        # A resumed run starts a FRESH frame stream and the output loop skips
+        # re-emitting checkpoint-restored vertices, so any terminal output the
+        # pre-pause pass produced would be lost. Pre-seed from the stash _suspend
+        # wrote to job_metadata; _dedup_outputs lets a re-emitted vertex overwrite
+        # its stale entry at finalize (resumed pass wins).
+        output_events: list[dict[str, Any]] = await self._load_pre_pause_outputs(job_id) if resume is not None else []
         async for frame_bytes, event_type in self._frame_source(**source_kwargs):
+            if event_type == WORKFLOW_OUTPUT_CAPTURE_EVENT:
+                # Off-wire terminal-output capture (both protocols): record it into
+                # the in-memory result ONLY. Deliberately no ``append_event`` (never
+                # in ``job_events``) and no ``publish`` (never on the live bus), so the
+                # wire is unchanged and both protocols populate ``Job.result``.
+                payload = self._decode_payload(frame_bytes)
+                output_data = payload.get("data")
+                if isinstance(output_data, dict):
+                    output_events.append(output_data)
+                continue
             if event_type == HUMAN_INPUT_REQUIRED_EVENT:
                 from langflow.services.jobs.service import _unwrap_pause_payload
 
                 payload = self._decode_payload(frame_bytes)
                 request = _unwrap_pause_payload(payload) or {}
-                raise PauseRequested(payload=payload, request_id=request.get("request_id"))
+                # Carry the pre-pause captures out with the exception so _suspend can
+                # stash them; without this the resumed pass starts empty and the final
+                # result drops every terminal output produced before the pause.
+                raise PauseRequested(
+                    payload=payload,
+                    request_id=request.get("request_id"),
+                    output_events=output_events,
+                )
             if self._adapter.is_durable(event_type):
                 # Vertex/milestone-boundary cooperative cancel: a STOP written to
                 # the durable signal table flips the job at the next durable
@@ -258,12 +287,14 @@ async def _drive(self, *, job_id: UUID, source_kwargs: dict[str, Any]) -> None:
                 seq = await self._jobs.append_event(job_id, event_type, payload)
                 last_durable_seq = seq
                 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":
+                if event_type == "output":
+                    # Fallback for custom frame sources that emit the standard
+                    # adapter event but not WORKFLOW_OUTPUT_CAPTURE_EVENT.
                     output_data = payload.get("data")
                     if isinstance(output_data, dict):
                         output_events.append(output_data)
+                if event_type == self._adapter.terminal_error_type:
+                    errored_payload = payload
             else:
                 await self._bus.publish(
                     str(job_id),
@@ -287,7 +318,42 @@ async def _drive(self, *, job_id: UUID, source_kwargs: dict[str, Any]) -> None:
             # 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", "outputs": output_events})
+        await self._jobs.set_result(job_id, {"status": "completed", "outputs": self._dedup_outputs(output_events)})
+
+    async def _load_pre_pause_outputs(self, job_id: UUID) -> list[dict[str, Any]]:
+        """Return terminal outputs a prior suspend stashed in job_metadata, or [].
+
+        Read once at the top of a resumed ``_drive`` to seed the capture list so
+        outputs produced before the pause survive into the completed-run result.
+        """
+        job = await self._jobs.get_job_by_job_id(job_id)
+        prior = (job.job_metadata or {}).get("pre_pause_outputs") if job else None
+        if isinstance(prior, list):
+            return [event for event in prior if isinstance(event, dict)]
+        return []
+
+    @staticmethod
+    def _dedup_outputs(events: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        """Merge pre-pause and resumed-pass captures keyed by ``component_id``.
+
+        A resumed run re-emits the branch vertices it had to rebuild, so one
+        ``component_id`` can appear both in the pre-seeded pre-pause outputs and in
+        the fresh captures. Keep each component's FIRST position (pre-pause outputs
+        stay ahead of ones produced after resume) but its LATEST value (the resumed
+        pass wins for anything it re-emitted). Events without a ``component_id`` are
+        passed through untouched, in order.
+        """
+        index: dict[str, int] = {}
+        merged: list[dict[str, Any]] = []
+        for event in events:
+            cid = event.get("component_id")
+            if cid is not None and cid in index:
+                merged[index[cid]] = event
+                continue
+            if cid is not None:
+                index[cid] = len(merged)
+            merged.append(event)
+        return merged
 
     async def _maybe_resume(self, job_id: UUID) -> dict[str, Any] | None:
         data = await self._pending_resume(job_id)

src/backend/base/langflow/services/jobs/exceptions.py

@@ -28,7 +28,17 @@ class PauseRequested(JobError):  # noqa: N818
     re-raises it without writing a terminal status.
     """
 
-    def __init__(self, payload: dict | None = None, request_id: str | None = None) -> None:
+    def __init__(
+        self,
+        payload: dict | None = None,
+        request_id: str | None = None,
+        output_events: list[dict] | None = None,
+    ) -> None:
         super().__init__("Run paused for human input")
         self.payload = payload or {}
         self.request_id = request_id
+        # Terminal outputs the run captured BEFORE the pause (e.g. a fully-run
+        # sibling branch ending in an output component). The runner stashes these
+        # on suspend so the resumed pass — which starts a fresh capture list and
+        # skips re-emitting already-built vertices — does not drop them.
+        self.output_events = output_events or []