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>

Author: 2 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,11 +648,35 @@ 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 durable ``output`` events
+            # the runner captured into ``Job.result`` (langflow-protocol). This
+            # needs no ``vertex_build`` rows, so it works with vertex-build storage
+            # off (headless), and skips the graph reconstruction the vertex-build
+            # path does.
+            result = job.result if isinstance(job.result, dict) else {}
+            output_events = result.get("outputs") or []
+            if output_events:
+                return 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 (an agui-protocol run
+            # leaves it empty; the terminal output lives only on /events) or a
+            # legacy job predates output capture. Reconstruct from the
+            # ``vertex_build`` rows keyed by job_id when they exist; if none do,
+            # ValueError degrades to a bare COMPLETED with an empty outputs map.
             try:
                 return await reconstruct_workflow_response_from_job_id(
                     session=session,
@@ -657,15 +685,11 @@ async def get_workflow_status(
                     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 {}
                 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,
                 )
 
         if job.status == JobStatus.FAILED:

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

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
@@ -239,7 +240,25 @@ async def _drive(self, *, job_id: UUID, source_kwargs: dict[str, Any]) -> None:
         # 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]] = []
+        # Durable event-log storage toggle. When off, the per-milestone job_events
+        # writes are skipped (reattach/replay via GET /events is unavailable); live
+        # streaming and the completed-run Job.result are unaffected. Read once so a
+        # mid-run settings flip cannot desync the seq monotonicity within this run.
+        from lfx.services.deps import get_settings_service
+
+        persist_events = get_settings_service().settings.job_events_storage_enabled
         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 ``Job.result`` fills independently of
+                # durable-event storage — even if job-event writing were disabled.
+                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
 
@@ -255,15 +274,22 @@ async def _drive(self, *, job_id: UUID, source_kwargs: dict[str, Any]) -> None:
                 if await self._stop_requested(job_id):
                     raise self._user_cancelled()
                 payload = self._decode_payload(frame_bytes)
-                seq = await self._jobs.append_event(job_id, event_type, payload)
+                if persist_events:
+                    seq = await self._jobs.append_event(job_id, event_type, payload)
+                else:
+                    # Event-log storage off: keep a local monotonic seq so live
+                    # subscribers still get ordered frames; there is no durable row
+                    # to read, so reattach/replay is intentionally unavailable.
+                    seq = last_durable_seq + 1
                 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":
-                    output_data = payload.get("data")
-                    if isinstance(output_data, dict):
-                        output_events.append(output_data)
+                # NOTE: terminal outputs are captured off-wire via the
+                # WORKFLOW_OUTPUT_CAPTURE_EVENT frame above (protocol-neutral), not
+                # from the langflow adapter's durable wire ``output`` event — that
+                # left agui-protocol ``Job.result`` empty. The wire ``output`` still
+                # flows to streaming clients via ``append_event``/``publish`` here.
             else:
                 await self._bus.publish(
                     str(job_id),

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

@@ -132,6 +132,166 @@ async def test_background_status_returns_output(client, created_api_key, bg_flow
     assert body["outputs"], f"completed background status carried no outputs: {body}"
 
 
+async def test_background_status_from_job_table_with_vertex_builds_off(client, created_api_key, bg_flow):
+    """With vertex_build storage OFF, GET status must still carry the full output.
+
+    Proves the headless-executor path: disable ``vertex_builds_storage_enabled`` so
+    NO vertex_build rows are written, run a background job, then assert the GET
+    status output is sourced from the durable ``Job.result`` blob (the fallback the
+    COMPLETED branch takes when vertex-build reconstruction finds nothing).
+
+    Three proofs: (1) zero vertex_build rows keyed by job_id, (2) ``Job.result``
+    holds the captured outputs, (3) GET status returns a non-empty ``outputs`` map.
+    """
+    from uuid import UUID
+
+    from langflow.services.database.models.jobs.model import Job, JobStatus
+    from langflow.services.database.models.vertex_builds.crud import get_vertex_builds_by_job_id
+    from lfx.services.deps import get_settings_service
+
+    settings = get_settings_service().settings
+    original = settings.vertex_builds_storage_enabled
+    settings.vertex_builds_storage_enabled = False
+    try:
+        submit = await client.post("api/v2/workflows", json=_body(bg_flow), headers=_headers(created_api_key))
+        assert submit.status_code == 200, submit.text
+        job_id = submit.json()["job_id"]
+
+        row = None
+        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 (
+                JobStatus.COMPLETED,
+                JobStatus.FAILED,
+                JobStatus.TIMED_OUT,
+            ):
+                break
+            await asyncio.sleep(0.1)
+        assert row is not None, "job row was never created"
+        assert row.status == JobStatus.COMPLETED, f"job did not complete: {row.status}"
+
+        # Proof 1: storage OFF => no vertex_build rows persisted for this job_id.
+        async with session_scope() as session:
+            vbs = await get_vertex_builds_by_job_id(session, job_id)
+        assert not vbs, f"vertex_builds were written despite storage OFF: {len(vbs)} rows"
+
+        # Proof 2: the durable Job.result blob carries the captured terminal outputs.
+        assert isinstance(row.result, dict), f"Job.result is not a dict: {row.result!r}"
+        assert row.result.get("outputs"), f"Job.result carried no outputs: {row.result}"
+
+        # Proof 3: GET status returns the full output, sourced from Job.result (reconstruct
+        # finds nothing with storage off, so the COMPLETED branch falls back to 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
+        body = status.json()
+        assert body["status"] == "completed"
+        assert body["outputs"], f"GET status carried no outputs with vertex_builds OFF: {body}"
+        # Proof 4: the Job.result path recovers the session_id from the terminal
+        # events (parity with the vertex-build path), so a background GET can
+        # continue the same chat thread even with vertex-build storage off.
+        assert body.get("session_id"), f"GET status lost session_id with vertex_builds OFF: {body}"
+    finally:
+        settings.vertex_builds_storage_enabled = original
+
+
+async def test_background_agui_populates_job_result_outputs(client, created_api_key, bg_flow):
+    """An agui-protocol background run now fills ``Job.result.outputs`` too.
+
+    Regression guard for the off-wire capture (WORKFLOW_OUTPUT_CAPTURE_EVENT): the
+    agui adapter emits no wire ``output`` event, so the runner used to leave
+    ``Job.result`` result-less and a GET status carried an empty ``outputs``. The
+    frame source now synthesizes a protocol-neutral capture frame from the raw
+    ``end_vertex``, which the runner records into ``Job.result`` without touching
+    ``job_events`` or the live bus. Proves: (1) ``Job.result`` carries outputs for
+    an agui run, (2) GET status returns a non-empty ``outputs`` map.
+    """
+    from uuid import UUID
+
+    from langflow.services.database.models.jobs.model import Job, JobStatus
+
+    body = {**_body(bg_flow), "stream_protocol": "agui"}
+    submit = await client.post("api/v2/workflows", json=body, headers=_headers(created_api_key))
+    assert submit.status_code == 200, submit.text
+    job_id = submit.json()["job_id"]
+
+    row = None
+    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 (JobStatus.COMPLETED, JobStatus.FAILED, JobStatus.TIMED_OUT):
+            break
+        await asyncio.sleep(0.1)
+    assert row is not None, "agui job row was never created"
+    assert row.status == JobStatus.COMPLETED, f"agui job did not complete: {row.status}"
+
+    # Proof 1: the durable Job.result blob carries the captured terminal outputs
+    # even though agui emitted no wire ``output`` event.
+    assert isinstance(row.result, dict), f"Job.result is not a dict: {row.result!r}"
+    assert row.result.get("outputs"), f"agui Job.result carried no outputs: {row.result}"
+
+    # Proof 2: GET status returns the full output, sourced 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
+    status_body = status.json()
+    assert status_body["status"] == "completed"
+    assert status_body["outputs"], f"agui GET status carried no outputs: {status_body}"
+
+
+async def test_background_with_job_events_storage_off(client, created_api_key, bg_flow):
+    """With job-event storage OFF, the run still completes and GET carries full output.
+
+    Proves the headless toggle: disable ``job_events_storage_enabled`` so the durable
+    milestone log is NOT written, run a background job, then assert (1) the run still
+    reaches COMPLETED, (2) no streaming ``job_events`` rows were persisted, (3) the
+    completed-run ``Job.result`` still carries outputs and GET status returns them —
+    because Job.result is a separate job-table write, independent of job_events.
+    """
+    from uuid import UUID
+
+    from langflow.services.database.models.jobs.model import Job, JobEvent, JobStatus
+    from lfx.services.deps import get_settings_service
+    from sqlalchemy import func, select
+
+    settings = get_settings_service().settings
+    original = settings.job_events_storage_enabled
+    settings.job_events_storage_enabled = False
+    try:
+        submit = await client.post("api/v2/workflows", json=_body(bg_flow), headers=_headers(created_api_key))
+        assert submit.status_code == 200, submit.text
+        job_id = submit.json()["job_id"]
+
+        row = None
+        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 (JobStatus.COMPLETED, JobStatus.FAILED, JobStatus.TIMED_OUT):
+                break
+            await asyncio.sleep(0.1)
+        assert row is not None, "job row was never created"
+        assert row.status == JobStatus.COMPLETED, f"job did not complete: {row.status}"
+
+        # Proof 1: no streaming milestone rows were written for this job.
+        async with session_scope() as session:
+            count = await session.scalar(
+                select(func.count()).select_from(JobEvent).where(JobEvent.job_id == UUID(job_id))
+            )
+        assert count == 0, f"job_events rows were written despite storage OFF: {count}"
+
+        # Proof 2: Job.result (separate job-table write) still carries outputs.
+        assert isinstance(row.result, dict), f"Job.result is not a dict: {row.result!r}"
+        assert row.result.get("outputs"), f"Job.result empty: {row.result}"
+
+        # Proof 3: GET status returns the full output, sourced 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
+        body = status.json()
+        assert body["status"] == "completed"
+        assert body["outputs"], f"GET status carried no outputs with job_events OFF: {body}"
+    finally:
+        settings.job_events_storage_enabled = original
+
+
 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.
 

src/lfx/src/lfx/services/settings/groups/telemetry.py

@@ -20,6 +20,14 @@ class TelemetrySettings(BaseModel):
     """If set to True, Langflow will track transactions between flows."""
     vertex_builds_storage_enabled: bool = True
     """If set to True, Langflow will keep track of each vertex builds (outputs) in the UI for any flow."""
+    job_events_storage_enabled: bool = True
+    """If set to True, background runs persist their durable event log to the ``job_events`` table.
+
+    This log powers the ``GET /api/v2/workflows/{job_id}/events`` reattach/replay (resume from
+    ``Last-Event-ID``). Set to False for a headless executor that never re-attaches to a run's stream:
+    the per-milestone DB writes are skipped and reattach/replay returns nothing. Live streaming while
+    connected and the completed-run ``Job.result`` (GET-status full output) are UNAFFECTED — those do
+    not depend on this table."""
 
     telemetry_writer_enabled: bool = True
     """Route transaction and vertex_build writes through an async batched writer backed by a