feat(lfx-serve): lfx owns the v2 workflow router via a host-DI seam (#13816)

* feat: native v2 workflows endpoint with pluggable stream protocols

Rebased onto release-1.10.0. The base independently rebuilt the v2
workflows backend (RBAC, body globals, share-aware fetch); keep our
forward design and conform its auth to that work:

1. Auth: keep get_current_user_for_workflow (session-or-API-key authN
   that does not hold a DB connection during the inline run, avoiding
   the SQLite lock contention api_key_security would cause) and enforce
   the base's RBAC on top: ensure_flow_permission(EXECUTE) before run,
   (READ) before status reconstruct, with widen_for_shares fetch.
2. Port the base's request-body globals onto the v2 WorkflowRunRequest.
   The X-LANGFLOW-GLOBAL-VAR-* headers stay supported (the Responses API
   passes globals that way); body globals win on conflict. Converters
   echo the effective globals via effective_globals.
3. Public endpoint keeps the v1 build_public_tmp posture
   (access_type==PUBLIC, run-as-owner); RBAC applies to the
   authenticated endpoint only.
4. Preserve the base's post-build KB-cache invalidation in the AG-UI
   build path.

The endpoint, AG-UI bridge, pluggable stream adapters, public endpoint,
and re-attach are unchanged.

* feat(api): add output_text and session_id to v2 workflow response

The synchronous /api/v2/workflows response keyed every result under its
component id, so reading the answer meant knowing an id you can't predict.
Surface two additive fields:

- output_text: the flow's single text answer (ChatOutput/TextOutput). None
  when the flow has zero or multiple text outputs, so callers read outputs
  rather than the shortcut guessing which channel is the answer.
- session_id: echoes the resolved session so chat/memory callers can
  continue the same thread (v1 /run returned this; v2 had dropped it).

outputs is unchanged, so this is non-breaking.

* test(api/v2): cover output_text and session_id on the v2 workflow response

Pin the sync-response shortcuts on the v2 workflows endpoint:
- output_text surfaces the lone ChatOutput/TextOutput text and stays None for
  non-output message nodes, data-only flows, and multi-text flows
- session_id echoes the resolved session; the error response exposes neither
- each outputs entry exposes only {type, status, content, metadata}, with the
  component id carried by the dict key

Also drop the component_id kwarg the converter passed to ComponentOutput, which
has no such field and silently dropped it.

* feat(api/v2): structured output with resolution reason on v2 response

Replace the flat output_text shortcut with an `output` object carrying the
resolved text answer plus a `reason` that explains why it resolved that way
(single/multiple/none/non_string/failed), so a null answer is always
diagnosable instead of silently None. `reason` follows the LLM-domain
finish_reason/stop_reason convention, distinct from the lifecycle status.

Also add `display_name` to each ComponentOutput (the stable component id
stays the dict key) and a computed `has_errors` flag derived from errors.

* feat(api/v2): add request-side output selection (output_ids)

Let a sync caller name the output(s) they want via output_ids so
output.text resolves deterministically (reason=single) on multi-output
flows instead of going null. Selection is steer-only: it picks the
answer among the named outputs without filtering the outputs map.

Invalid ids are rejected with 422 before the flow runs (and before any
job row is created), so a typo costs no compute. Resolution considers
selected outputs that actually fired, so branching flows resolve to
whichever candidate ran.

* feat(api/v2): emit per-output events on the langflow stream

Give v2-workflows sync and the langflow stream protocol one parser. The
stream now emits a normalized "output" event per terminal output carrying
an OutputEvent (the ComponentOutput shape sync returns in outputs[id], plus
component_id). A shared build_component_output() backs both the sync
converter and the adapter, and the build loop ships authoritative vertex
metadata as an additive output_meta key on end_vertex (existing consumers
read build_data and ignore it).

This is access-pattern parity (one parser, same fields, same terminal set),
not byte-identical content: the stream reuses the v1 build path whose
display serialization differs from sync's run_graph output.

* fix(api/v2): enforce no-code-execution gate on public workflow endpoint

The v2 public endpoint only ran validate_flow_for_current_settings and
skipped validate_public_flow_no_code_execution, which the v1
build_public_tmp path applies. A public flow containing a Python
interpreter/REPL (or the legacy Python Code Structured tool, Smart
Transform lambda) was therefore an unauthenticated server-side
code-execution primitive (report H1-3754930).

Mirror v1: import the validator and call it right after the
public-access gate. PublicFlowValidationError subclasses
CustomComponentValidationError, so the existing handler already
sanitizes it to a 400 'This flow cannot be executed.' without leaking
the blocked component class names.

Add a non-mocking test that builds a public flow with a real
PythonREPLComponent and asserts the sanitized 400 (verified RED: returns
200 without the gate).

LE-1389

* fix(api/v2): reconstruct background workflow status from job-keyed vertex builds

A completed background job's GET status 500'd with 'No vertex builds found
for job_id'. The background build path differed from the sync path twice:

1. generate_flow_events minted a fresh run_id instead of using job_id, so
   vertex builds were keyed by an id the status query never uses. Thread
   run_id through _stream_event_frames -> generate_flow_events and pass
   job_id from the background buffer so graph.run_id == job_id (the sync
   path already does graph.set_run_id(job_id)).

2. The SSE build loop (build_vertices) only persisted builds when log_builds
   was set and never passed job_id. Tie log_builds to job-tracked runs
   (run_id present) and pass job_id=graph.run_id on the persist call.
   Job-tracked runs also persist streaming terminal vertices so
   reconstruction is complete; the live build path (run_id is None) keeps
   its original behavior, so the v1 build path is unchanged.

Test: a real background run polled to completion, then GET status asserts a
reconstructed 200 (verified RED: 500 'No vertex builds found' before the
fix). Covers the non-streaming flow. v1 build path unchanged (35 build
tests pass); AG-UI suite 46 pass.

LE-1389

* fix(api/v2): merge workflow AG-UI cancellation hardening LE-1389

* fix(api/v2): signal cross-worker workflow stops LE-1389

* fix(api/v2): report unconfirmed workflow stops LE-1389

* fix(api/v2): keep background workflows out of polling watchdog LE-1389

* fix(api/v2): buffer parallel messages in the AG-UI translator instead of dropping them

Parallel components stream tokens for different message ids interleaved.
The translator tracked a single open message: the first foreign token
closed the open message and tombstoned its id, so every later event for
it was dropped and its remaining text never reached the client.

Tokens for a message that cannot take the wire now buffer until the open
message genuinely ends (its add_message finalizer), then flush in arrival
order; complete messages landing mid-stream buffer the same way instead
of interleaving a second START. end/error drain all buffers before the
terminal event. The wire still carries at most one open text message, so
the stream stays AG-UI-conformant.

* fix(api/v2): gate AG-UI message finalization on non-partial state and purge removed buffers

A partial add_message re-fire (the agent path emits these at tool
start/end for a message it is still streaming) was treated as the
finalizer: it closed and tombstoned the id, so the post-tool answer
was dropped. Only a non-partial add_message finalizes now; state
defaults to complete, so payloads without properties are unchanged.

remove_message now purges a buffered message and tombstones its id,
so text the backend retracted is not flushed to the client later.

* Fix AG-UI workflow lifecycle edges

* [autofix.ci] apply automated fixes

* fix(frontend): enable downlevelIteration for jest Set/Map iteration

ts-jest compiles with target es5; without downlevelIteration, [...set] and
for...of over a Set/Map emit ES5 that yields nothing. That silently broke the
AG-UI bridge tests: runningNodeIds spread, markRunningNodesFailed, and
restoreOriginalBuildStatuses all iterated empty. Production (Vite/SWC, modern
target) was never affected; only the ts-jest harness was. Fixes the 3 failing
jest tests on this branch with no other suite changes (4994/4994 pass).

* fix(api/v2): surface inactivated branch vertices over AG-UI

A branch component (If-Else, Conditional Router) reports its not-taken
vertices in build_data.inactivated_vertices, but the AG-UI translator only
emitted the branch node's own success/error status and dropped that list. The
canvas seeds every planned node as pending from vertices_sorted; skipped
vertices then get no build_start/end_vertex, so they stayed stuck on pending
instead of rendering as inactive (the v1 build path marked them INACTIVE).

The translator now appends an inactive STATE_DELTA op per inactivated vertex,
and the frontend bridge maps the new inactive status to BuildStatus.INACTIVE
and tears its edges down like a completed node. Fixes the If-Else regression
in general-bugs-reset-flow-run.spec.ts.

* fix(api/v2): dedupe repeated inactive node deltas in AG-UI stream

build.py keeps reporting a conditionally-excluded vertex in
inactivated_vertices on every subsequent end_vertex (the excluded set
persists until the ConditionalRouter clears it), so the translator was
putting the same inactive STATE_DELTA on the wire once per remaining
vertex. Track emitted inactive nodes and skip re-emitting; drop a node
from the set when it actually runs again (build_start/end_vertex) so a
loop re-activation can still re-emit inactive later.

* fix(api/v2): address review findings on the v2 workflows endpoint

- recover session_id for completed background jobs from the persisted
  terminal message instead of always returning null, so GET status can
  continue the same chat/memory thread
- replay a user-cancel as a CUSTOM cancel marker + RUN_FINISHED (agui) and
  a `cancelled` terminal (langflow) instead of RUN_ERROR, so a re-attaching
  client no longer reads a deliberate stop as a failure
- cancel the evicted still-running buffer writer when the background-run
  registry is full, so it stops appending into a run no reader can find
- derive per-component status from the error artifact / valid flag instead
  of hardcoding COMPLETED, and stop the langflow adapter dropping `valid`
- throttle the unauthenticated public endpoint per IP and bound its
  input_value/session_id length
- document the sync-only scope of request-body globals
- document that live event re-attach is intentionally owner-only

* test(lfx): register public_flow_rate_limit_per_minute in settings composition

* refactor(v2 workflows): split workflow.py and address review blockers

Splits the ~1.5k-line workflow.py into focused modules and folds in the
execution-timeout and error-sanitization fixes from Cristhianzl's review of #13307.

- B1: workflow.py now holds only the four route handlers. Validation guards move
  to workflow_validation, the sync/stream run loop to workflow_execution, and the
  durable background machinery to workflow_background (layered, acyclic).
- I1: add workflow_execution_timeout (default 300) and apply a single wall-clock
  ceiling across sync, stream, background, and public via _stream_event_frames. A
  timeout becomes a sanitized terminal error and marks a background job failed.
- I3: the route error handlers no longer echo raw exception text. They return a
  generic, code-tagged message and log the full exception server-side.
- R1: remove the "commented out / future scope" comments that sat over live
  dataframe-extraction code in converters.py.
- R4: drop the worker-routing internals from the reattach 409 message.

Tests cover the timeout terminal-error path and the error-body sanitization, and
the settings field-count guard is updated for the new setting.

* refactor(lfx): extract v2 workflow contract layer into lfx.workflow

Moves the protocol-agnostic pieces of the v2 workflows API out of the langflow
backend into lfx so both the backend and `lfx serve` can share one contract.
First step toward giving lfx (the production runtime) the v2 workflows API.

- Move api/v2/adapters/, agui_translator.py, and converters.py to lfx/workflow/.
  They depend only on lfx.schema.workflow and ag_ui (already an lfx dep), so lfx
  carries the contract with zero langflow imports.
- Decouple the one langflow reference: converters typed run_response against
  langflow.api.v1.schemas.RunResponse (TYPE_CHECKING only). Replaced with a local
  RunResponseLike Protocol (outputs + session_id), the only attributes used.
- Repoint the six backend v2 workflow modules to import from lfx.workflow.
- Move the five protocol-agnostic contract tests into src/lfx/tests/unit/workflow/
  (run in the lfx-only env). test_output_event_parity and test_workflow_agui stay
  in langflow (they need langflow.api.build) with repointed imports.

Coverage unchanged: 201 contract tests pass in the lfx-only env, 191 backend v2
tests pass; 392 total, same as before the move.

* feat(lfx serve): v2 workflow endpoints (sync + stream) on lfx serve

Gives the production runtime (lfx serve) the same v2 contract as the langflow
backend's POST /api/v2/workflows, built on the shared lfx.workflow layer.

- New POST /workflows endpoint: WorkflowRunRequest in, WorkflowExecutionResponse
  (sync) or an SSE stream (langflow/agui protocols) out. flow_id resolves against
  the serve registry; per-request deepcopy+stamp mirrors the run/stream endpoints.
- sync runs via run_graph_internal (the same primitive the backend sync path
  uses) so the converter sees the aggregated RunOutputs shape.
- stream drives the run with a token-stream EventManager wired into
  execute_graph_with_capture and feeds queue events through the shared
  StreamAdapter; emits a terminal end so the adapter closes the run cleanly.
- background and public modes are rejected (422); tweaks/data/files/globals and
  partial-run boundaries are rejected too (no per-request graph rebuild yet).
- execute_graph_with_capture gains an optional event_manager param.

Tests build a real ChatInput->ChatOutput flow (no mocks) and exercise sync, both
stream protocols, and the guards. 329 lfx serve + contract tests pass.

* feat(lfx): workflow-router seam owned by lfx via a host-DI seam

lfx now exports a WorkflowHost Protocol + create_workflow_router() factory.
The router owns the env-neutral handler body (run dispatch, single SSE loop,
error mapping, dev-api guard); the host supplies auth/flow-lookup/session.
Bare serve uses a no-db ServeWorkflowHost (supports_background=False), which
collapses the duplicated serve handler into the shared router.

Background/durable routes register only when host.supports_background is True,
so bare serve's surface stays exactly POST /workflows. developer_api_guard is a
factory flag (default True) so serve keeps its current open behavior.

Adds a no-mock cross-host contract test pinning the request/response and SSE
shape so the two hosts can't drift.

* feat(api): route LF v2 workflow run through the lfx host-DI seam

Langflow now mounts the shared lfx workflow router for POST /api/v2/workflows
via a LangflowWorkflowHost (auth->UserRead in its own session, flow lookup,
ensure_flow_permission, readonly session, background submit). The durable
status/stop/events routes stay LF-rich on an LF-owned background router, so
only one handler exists per method+path. The public workflow router is
untouched.

LF streaming and sync stay LF-specific via host.run_sync / host.stream_response
seams (LF keeps the v1 build loop with agui side-channel + vertex persistence;
bare serve keeps lfx's lean defaults). create_workflow_router gains
auto_register_job_routes so LF can enable background submit without the router
auto-registering its generic job routes.

* fix(api): drop orphaned execute_workflow and keep stream-protocol precedence

Self-review follow-ups on the workflow-router seam:

- The shared router validated stream_protocol after fetching/authorizing the
  flow, so an unknown protocol against a missing/unauthorized flow returned
  404/403 instead of the old 422. Move the 422 check above get_flow to keep the
  pre-seam precedence.
- execute_workflow lost its route decorator when the run path moved to the lfx
  router via LangflowWorkflowHost, leaving a production-dead function kept only
  for three direct-call tests. Delete it and point those tests at the real
  helpers the host calls (build_stream_response, authorize_flow_action), which
  also drops their get_flow_by_id_or_endpoint_name mocks.

* fix(api): echo requested flow identifier in v2 workflow 404 reframes

A denial or owner-override 404 echoed str(flow.id) (the resolved internal
UUID). When the caller referenced the flow by endpoint name, that leaked the
canonical UUID and changed the flow_id in the error body vs the request. The
pre-seam handler echoed the requested identifier; restore that.

authorize_flow_action takes the requested id (LangflowWorkflowHost passes
ResolvedFlow.flow_id, which already holds the caller's value); the owner-gate
reframe uses parsed.flow_id. Adds a test that a denied endpoint-name request
echoes the name, not the UUID. Also documents ResolvedFlow.graph as a
host-defined artifact (Graph for serve, FlowRead for langflow), per review.

* test+docs: pin precedence and leak fixes, restore POST OpenAPI schema

From the ultracode review:
- The 422 stream-protocol precedence was untested (all cases used an existing
  flow). Pin it: a host whose get_flow 404s plus a bad protocol must still 422.
- The UUID-leak test exercised the helper directly, not the host wiring that
  supplies requested_id. Route it through LangflowWorkflowHost.authorize so a
  regression in that wiring is caught.
- The authenticated POST lost its responses= OpenAPI schema when the route
  moved to the shared router. Thread responses through create_workflow_router
  and restore WORKFLOW_EXECUTION_RESPONSES on the langflow mount.
- Soften the router docstring: SSE framing is single-sourced only for hosts on
  the lfx default (langflow overrides stream_response).

* fix(v2): restore OperationalError->503 on the inline-run and background paths; correct router developer_api_guard docstring

A DB OperationalError raised during the run (create_job runs outside the inner
try) fell through to a 500 instead of the pre-seam 503 DATABASE_ERROR on both
run_sync_with_mapping and submit_background_with_mapping. Add the missing
OperationalError->503 branch (body byte-identical to the fetch-path mapping).

Also fix the create_workflow_router docstring: it claimed langflow keeps
developer_api_guard=True, but every mount passes False and the v2 surface never
carried the guard.

* fix(lfx-serve): align v2 workflow endpoints with backend contract

Accept request-level globals (applied as request-scoped variables) instead
of rejecting them with 422; validate output_ids against the flow's terminal
nodes before running (422 on unknown) rather than wasting a run; and convert
stream-queue overflow into an explicit error frame instead of silently
dropping events for a slow SSE client.

* fix(lfx-serve): carry the v2 contract fixes into the router seam

The seam moved the bare-serve endpoint logic out of serve_workflow.py into
lfx.workflow.router before three contract fixes landed on the inline version,
so router.py's default path was missing them. Port them into the lfx-default
run/stream functions (bare serve only; the langflow host overrides these):

- accept request-level globals, applied as request-scoped variables on the
  per-request graph copy, and echoed as effective_globals (drop the globals
  rejection in _reject_unsupported_fields)
- reject unknown output_ids up front with 422 UNKNOWN_OUTPUT_IDS
- replace the plain bounded asyncio.Queue with _WorkflowEventQueue so a slow
  SSE client gets an explicit error frame instead of silently dropped events

Repoint the serve test's _terminal_node_ids/_WorkflowEventQueue import at
lfx.workflow.router (they moved out of serve_workflow.py).

* fix: honor no_env_fallback on sync runs and map DB errors in authorize

Two points from Debojit's review:

- Sync /workflows bypassed the no_env_fallback isolation. run_workflow_sync
  activated request_variables but not the no_env_fallback contextvar, so a sync
  run under LFX_SERVE_NO_ENV_FALLBACK=1 still resolved credentials from
  os.environ while stream (via execute_graph_with_capture) was isolated. Activate
  it around the run and reset in finally, mirroring the stream path.

- authorize_flow_action caught only HTTPException, so an OperationalError from
  ensure_flow_permission's audit write (DB lock under inline-run contention)
  escaped as a bare 500 instead of the retryable 503 DATABASE_ERROR contract the
  fetch path keeps. Add the same OperationalError -> 503 / Exception -> 500 arms.

Tests: a no-mock real-graph test that the sync path activates the isolation
mid-run and resets after, and a 503-on-DB-lock test through the host wiring.

* fix(lfx-serve): mount v2 workflows at /api/v2/workflows to match backend

Per Debojit's review: serve exposed the shared workflow router at /workflows
while the langflow backend serves it at /api/v2/workflows, so switching runtimes
meant changing the URL path, not just the host. Mount the serve router under
/api/v2 so the path is identical across runtimes; a client points at a different
host/environment with no path change. serve's other routes (/flows, /health)
stay root-level. Updates the serve integration tests and docstring.

* test(lfx-serve): handle FastAPI >=0.137 lazy _IncludedRouter in serve route assertions

release-1.11.0's FastAPI bump makes include_router mount the /api/v2 workflow
router as a lazy _IncludedRouter wrapper with no .path, so the three
create_serve_app route-introspection tests hit AttributeError. Skip wrappers
(these assert directly-mounted serve paths only), matching how the langflow
backend already handles the same >=0.137 behavior.

---------

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

Author: ogabrielluiz

.secrets.baseline

@@ -197,7 +197,7 @@
         "filename": ".github/workflows/release_nightly.yml",
         "hashed_secret": "3e26d6750975d678acb8fa35a0f69237881576b0",
         "is_verified": false,
-        "line_number": 601,
+        "line_number": 750,
         "is_secret": false
       }
     ],
@@ -6687,7 +6687,7 @@
         "filename": "src/lfx/src/lfx/cli/serve_app.py",
         "hashed_secret": "b894b81be94cf8fa8d7536475aaec876addf05c8",
         "is_verified": false,
-        "line_number": 59,
+        "line_number": 60,
         "is_secret": false
       }
     ],
@@ -7272,5 +7272,5 @@
       }
     ]
   },
-  "generated_at": "2026-07-07T22:31:52Z"
+  "generated_at": "2026-07-08T10:30:03Z"
 }

src/backend/base/langflow/api/router.py

@@ -1,6 +1,9 @@
 # Router for base api
 from fastapi import APIRouter
+from lfx.schema.workflow import WORKFLOW_EXECUTION_RESPONSES
 from lfx.services.settings.feature_flags import FEATURE_FLAGS
+from lfx.workflow.host import WorkflowHost
+from lfx.workflow.router import create_workflow_router
 
 from langflow.api.v1 import (
     api_key_router,
@@ -39,8 +42,9 @@
 from langflow.api.v2 import files_router as files_router_v2
 from langflow.api.v2 import mcp_router as mcp_router_v2
 from langflow.api.v2 import registration_router as registration_router_v2
+from langflow.api.v2 import workflow_background_router as workflow_background_router_v2
 from langflow.api.v2 import workflow_public_router as workflow_public_router_v2
-from langflow.api.v2 import workflow_router as workflow_router_v2
+from langflow.api.v2.workflow_host import LangflowWorkflowHost
 
 router_v1 = APIRouter(
     prefix="/v1",
@@ -124,7 +128,26 @@ def _include_agentic_router():
 router_v2.include_router(files_router_v2)
 router_v2.include_router(mcp_router_v2)
 router_v2.include_router(registration_router_v2)
-router_v2.include_router(workflow_router_v2)
+
+# POST /api/v2/workflows runs through the shared lfx router bound to the
+# langflow host. ``supports_background=True`` lets the background-submit branch
+# dispatch to the host; ``auto_register_job_routes=False`` suppresses the lfx
+# generic GET-status/POST-stop routes so the langflow durable router below owns
+# them (one handler per method+path). ``developer_api_guard=False`` because the
+# authenticated langflow v2 router has never carried a developer-api gate; the
+# default-off setting would otherwise 403 every authenticated request.
+_workflow_host = LangflowWorkflowHost()
+assert isinstance(_workflow_host, WorkflowHost)  # noqa: S101
+router_v2.include_router(
+    create_workflow_router(
+        _workflow_host,
+        developer_api_guard=False,
+        auto_register_job_routes=False,
+        responses=WORKFLOW_EXECUTION_RESPONSES,
+    )
+)
+# The langflow-owned durable routes: GET status, POST /stop, GET /{job_id}/events.
+router_v2.include_router(workflow_background_router_v2)
 router_v2.include_router(workflow_public_router_v2)
 
 router = APIRouter(

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

@@ -3,13 +3,15 @@
 from .files import router as files_router
 from .mcp import router as mcp_router
 from .registration import router as registration_router
-from .workflow import router as workflow_router
+from .workflow import router as workflow_background_router
+from .workflow_host import LangflowWorkflowHost
 from .workflow_public import router as workflow_public_router
 
 __all__ = [
+    "LangflowWorkflowHost",
     "files_router",
     "mcp_router",
     "registration_router",
+    "workflow_background_router",
     "workflow_public_router",
-    "workflow_router",
 ]