feat: native v2 workflows endpoint with pluggable stream protocols

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

@@ -55,6 +55,32 @@
 STREAMING_ACTIVITY_REFRESH_S = 10.0
 
 
+def _output_meta_for_vertex(graph: Graph, vertex_id: str) -> dict:
+    """Authoritative per-output metadata for the v2 ``output`` stream event.
+
+    Sourced from the real graph vertex (the only place ``display_name`` /
+    ``is_output`` / declared output types are authoritative) so the streamed
+    ``OutputEvent`` matches the sync ``outputs[id]`` ``ComponentOutput``. Shipped as
+    an additive ``output_meta`` key on ``end_vertex``; existing consumers read
+    ``build_data`` and ignore this. ``is_terminal`` mirrors the sync ``outputs`` set
+    so the stream emits an ``output`` event for exactly the same components.
+    """
+    vertex = graph.get_vertex(vertex_id)
+    output_types = vertex.outputs[0].get("types", []) if (vertex.outputs and len(vertex.outputs) > 0) else []
+    try:
+        terminal_ids = set(graph.get_terminal_nodes())
+    except AttributeError:
+        terminal_ids = {v.id for v in graph.vertices if not graph.successor_map.get(v.id, [])}
+    return {
+        "component_id": vertex.id,
+        "display_name": vertex.display_name or vertex.vertex_type,
+        "vertex_type": vertex.vertex_type,
+        "is_output": bool(vertex.is_output),
+        "is_terminal": vertex_id in terminal_ids,
+        "output_types": output_types,
+    }
+
+
 def _log_component_input_telemetry(
     vertex,
     vertex_id: str,
@@ -337,13 +363,19 @@ async def generate_flow_events(
     current_user: CurrentActiveUser,
     flow_name: str | None = None,
     source_flow_id: uuid.UUID | None = None,
+    run_id: str | None = None,
+    track_job_status: bool = True,
 ) -> None:
     """Generate events for flow building process.
 
     This function handles the core flow building logic and generates appropriate events:
     - Building and validating the graph
     - Processing vertices
     - Handling errors and cleanup
+
+    When ``run_id`` is provided the graph adopts it instead of minting a fresh
+    one, so callers (e.g. background jobs) can later look up the run's vertex
+    builds by that id. Defaults to a fresh uuid for the live build path.
     """
     chat_service = get_chat_service()
     telemetry_service = get_telemetry_service()
@@ -354,14 +386,14 @@ async def build_graph_and_get_order() -> tuple[list[str], list[str], Graph]:
         start_time = time.perf_counter()
         components_count = 0
         graph = None
-        run_id = str(uuid.uuid4())
+        build_run_id = run_id or str(uuid.uuid4())
         try:
             flow_id_str = str(flow_id)
             # Create a fresh session for database operations
             async with session_scope() as fresh_session:
                 graph = await create_graph(fresh_session, flow_id_str, flow_name)
 
-            graph.set_run_id(run_id)
+            graph.set_run_id(build_run_id)
             first_layer = sort_vertices(graph)
 
             for vertex_id in first_layer:
@@ -374,13 +406,13 @@ async def build_graph_and_get_order() -> tuple[list[str], list[str], Graph]:
             vertices_to_run = list(graph.vertices_to_run.union(get_top_level_vertices(graph, graph.vertices_to_run)))
 
             await chat_service.set_cache(flow_id_str, graph)
-            await log_telemetry(start_time, components_count, run_id=run_id, success=True)
+            await log_telemetry(start_time, components_count, run_id=build_run_id, success=True)
 
         except Exception as exc:
             await log_telemetry(
                 start_time,
                 components_count,
-                run_id=run_id,
+                run_id=build_run_id,
                 success=False,
                 error_message=str(exc),
             )
@@ -497,8 +529,12 @@ async def _build_vertex(vertex_id: str, graph: Graph, event_manager: EventManage
 
             result_data_response.message = artifacts
 
-            # Log the vertex build
-            if not vertex.will_stream and log_builds:
+            # Log the vertex build. Job-tracked runs (background workflows pass a
+            # ``run_id``) persist every vertex, including streaming terminal outputs,
+            # so GET-status reconstruction by job_id is complete. The live build
+            # path (``run_id is None``) keeps the original "skip streaming vertices"
+            # behavior unchanged.
+            if log_builds and (run_id is not None or not vertex.will_stream):
                 background_tasks.add_task(
                     log_vertex_build,
                     flow_id=flow_id_str,
@@ -507,6 +543,9 @@ async def _build_vertex(vertex_id: str, graph: Graph, event_manager: EventManage
                     params=params,
                     data=result_data_response,
                     artifacts=artifacts,
+                    # Key the persisted build by the run id so job-tracked runs can
+                    # reconstruct status by job_id.
+                    job_id=graph.run_id,
                 )
             else:
                 await chat_service.set_cache(flow_id_str, graph)
@@ -615,7 +654,9 @@ async def build_vertices(
             msg = f"Error serializing vertex build response: {exc}"
             raise ValueError(msg) from exc
 
-        event_manager.on_end_vertex(data={"build_data": build_data})
+        event_manager.on_end_vertex(
+            data={"build_data": build_data, "output_meta": _output_meta_for_vertex(graph, vertex_id)}
+        )
 
         if vertex_build_response.valid and vertex_build_response.next_vertices_ids:
             tasks = []
@@ -648,7 +689,7 @@ async def build_vertices(
     _build_run_id: uuid.UUID | None = None
     try:
         _build_run_id = uuid.UUID(graph.run_id) if graph.run_id else None
-        if _build_run_id is not None:
+        if track_job_status and _build_run_id is not None:
             _build_job_svc = get_job_service()
             await _build_job_svc.create_job(
                 job_id=_build_run_id,

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

@@ -39,6 +39,7 @@
 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_public_router as workflow_public_router_v2
 from langflow.api.v2 import workflow_router as workflow_router_v2
 
 router_v1 = APIRouter(
@@ -124,6 +125,7 @@ def _include_agentic_router():
 router_v2.include_router(mcp_router_v2)
 router_v2.include_router(registration_router_v2)
 router_v2.include_router(workflow_router_v2)
+router_v2.include_router(workflow_public_router_v2)
 
 router = APIRouter(
     prefix="/api",

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

@@ -45,6 +45,7 @@
     build_graph_from_db_no_cache,
     cascade_delete_flow,
     scope_session_to_namespace,
+    validate_public_files,
     verify_public_flow_and_get_user,
 )
 
@@ -90,5 +91,6 @@
     "remove_api_keys",
     "scope_session_to_namespace",
     "validate_is_component",
+    "validate_public_files",
     "verify_public_flow_and_get_user",
 ]

src/backend/base/langflow/api/utils/flow_utils.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import re
 import uuid
 from typing import TYPE_CHECKING
 
@@ -131,6 +132,42 @@ async def cascade_delete_flow(session: AsyncSession, flow_id: uuid.UUID) -> None
         raise RuntimeError(msg, e) from e
 
 
+# Public flow file paths must be ``{source_flow_id}/{safe_basename}`` — uploads
+# under that namespace are the only legitimate inputs for an unauthenticated
+# build. Anything else (absolute paths, traversal, foreign flow_ids) is a
+# probe at the arbitrary-file-read class of bug (GHSA-rcjh-r59h-gq37).
+_PUBLIC_FILE_PATH_RE = re.compile(
+    r"^([0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12})/([^/\\]+)$"
+)
+_PUBLIC_FILE_REJECTED_SUBSTRINGS = ("\x00", "..", "\\")
+
+
+def validate_public_files(files: list[str] | None, source_flow_id: uuid.UUID) -> None:
+    """Reject file references that aren't ``{source_flow_id}/{basename}``.
+
+    Mitigates GHSA-rcjh-r59h-gq37: an unauthenticated build must not be
+    able to address files outside its own flow's storage namespace.
+    Called from any endpoint that accepts caller-supplied file references
+    under a public-access boundary.
+    """
+    if not files:
+        return
+    expected_flow_id = str(source_flow_id).lower()
+    for entry in files:
+        if not isinstance(entry, str) or not entry:
+            raise HTTPException(status_code=400, detail="Invalid file entry")
+        if any(token in entry for token in _PUBLIC_FILE_REJECTED_SUBSTRINGS):
+            raise HTTPException(status_code=400, detail="Invalid file path")
+        match = _PUBLIC_FILE_PATH_RE.match(entry)
+        if not match:
+            raise HTTPException(status_code=400, detail="Invalid file path format")
+        flow_id_segment, basename = match.group(1), match.group(2)
+        if flow_id_segment.lower() != expected_flow_id:
+            raise HTTPException(status_code=400, detail="File not in this flow's namespace")
+        if basename in (".", ".."):
+            raise HTTPException(status_code=400, detail="Invalid filename")
+
+
 def compute_virtual_flow_id(identifier: str | uuid.UUID, flow_id: uuid.UUID) -> uuid.UUID:
     """Compute a deterministic virtual flow ID for session/message isolation.
 

src/backend/base/langflow/api/v1/chat.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import asyncio
-import re
 import time
 import traceback
 import uuid
@@ -34,6 +33,7 @@
     get_top_level_vertices,
     parse_exception,
     scope_session_to_namespace,
+    validate_public_files,
     verify_public_flow_and_get_user,
 )
 from langflow.api.v1.schemas import (
@@ -721,34 +721,9 @@ async def build_flow_and_stream(flow_id, inputs, background_tasks, current_user)
     )
 
 
-# Public flow file paths must be `{source_flow_id}/{safe_basename}` — uploads
-# under that namespace are the only legitimate inputs for an unauthenticated
-# build. Anything else (absolute paths, traversal, foreign flow_ids) is a
-# probe at the arbitrary-file-read class of bug.
-_PUBLIC_FILE_PATH_RE = re.compile(
-    r"^([0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12})/([^/\\]+)$"
-)
-_PUBLIC_FILE_REJECTED_SUBSTRINGS = ("\x00", "..", "\\")
-
-
-def _validate_public_files(files: list[str] | None, source_flow_id: uuid.UUID) -> None:
-    """Reject file references that aren't `{source_flow_id}/{basename}`."""
-    if not files:
-        return
-    expected_flow_id = str(source_flow_id).lower()
-    for entry in files:
-        if not isinstance(entry, str) or not entry:
-            raise HTTPException(status_code=400, detail="Invalid file entry")
-        if any(token in entry for token in _PUBLIC_FILE_REJECTED_SUBSTRINGS):
-            raise HTTPException(status_code=400, detail="Invalid file path")
-        match = _PUBLIC_FILE_PATH_RE.match(entry)
-        if not match:
-            raise HTTPException(status_code=400, detail="Invalid file path format")
-        flow_id_segment, basename = match.group(1), match.group(2)
-        if flow_id_segment.lower() != expected_flow_id:
-            raise HTTPException(status_code=400, detail="File not in this flow's namespace")
-        if basename in (".", ".."):
-            raise HTTPException(status_code=400, detail="Invalid filename")
+# NOTE: ``validate_public_files`` (the canonical helper that mitigates
+# GHSA-rcjh-r59h-gq37) was moved to ``langflow.api.utils.flow_utils`` so v2's
+# public workflow endpoint shares the exact same gate. Keep it imported above.
 
 
 @router.post("/build_public_tmp/{flow_id}/flow")
@@ -811,7 +786,7 @@ async def build_public_tmp(
         # Reject caller-supplied file references that aren't scoped to this
         # public flow's own storage namespace. Done before any flow lookup so
         # malformed requests fail fast and don't touch the DB.
-        _validate_public_files(files, flow_id)
+        validate_public_files(files, flow_id)
 
         # Verify this is a public flow and get the associated user
         client_id = request.cookies.get("client_id")

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

@@ -4,5 +4,12 @@
 from .mcp import router as mcp_router
 from .registration import router as registration_router
 from .workflow import router as workflow_router
+from .workflow_public import router as workflow_public_router
 
-__all__ = ["files_router", "mcp_router", "registration_router", "workflow_router"]
+__all__ = [
+    "files_router",
+    "mcp_router",
+    "registration_router",
+    "workflow_public_router",
+    "workflow_router",
+]