feat: native v2 workflows endpoint with pluggable stream protocols

[ogabrielluiz]

Routes the playground and canvas through POST /api/v2/workflows, a Langflow-native endpoint with a pluggable stream-protocol layer. Frontend pins stream_protocol: "agui" so the AG-UI typed event stream keeps driving canvas + chat-view. flowStore.buildFlow now goes through the v2 endpoint unconditionally; the frontend v1 build branch is removed. The backend api/v1/chat.py build pipeline still exists for now (retiring it is a follow-up below).

On the diff size

It's large on paper, but ~2/3 of it is tests: 5,438 of the 8,173 added lines (19 test files). The production surface is ~2,600 lines, and most of it lands in new, additions-only v2 modules. Edits to existing files are small (flowStore.ts is a net deletion of the v1 build path). Suggested read order: lfx/schema/workflow.py → api/v2/workflow.py → api/v2/adapters/ → converters.py / agui_translator.py → frontend controllers/API/agui/run-agent.ts + run-flow-bridge.ts → stores/flowStore.ts.

Backend

The endpoint takes a native WorkflowRunRequest (extra="forbid"):

{
  "flow_id": "...",
  "input_value": "...",
  "mode": "sync" | "stream" | "background",
  "stream_protocol": "langflow" | "agui",
  "session_id": "...",
  "tweaks": {...},
  "data": {"nodes": [...], "edges": [...]},
  "files": [...]
}

mode defaults to sync (one curl, one JSON back). stream_protocol defaults to langflow (passthrough of EventManager events as {"event": "...", "data": {...}}). Unknown stream_protocol returns 422 with the available list for every mode, not just streaming.

The sync (and status) response is flat. The text reply is surfaced at the top level as output_text (the flow's single ChatOutput/TextOutput; null when a flow has no single text output, so callers read outputs instead of the shortcut guessing which channel is the answer), and session_id echoes the resolved session so chat/memory callers can continue the same thread (v1 /run returned this; v2 had dropped it). The per-component results stay in outputs, keyed by component id. Both fields are additive, so the existing outputs contract is unchanged.

Stream dispatch goes through a StreamAdapter registry. langflow is passthrough; agui wraps the existing AGUITranslator. Adding a third protocol is a register_stream_adapter call.

Background mode buffers the chosen protocol's frames per job. GET /api/v2/workflows/{job_id}/events re-attaches with Last-Event-ID. /stop releases the buffer and wakes waiters so they see a clean stream end instead of hanging. The in-memory registry prefers evicting completed runs over still-running ones so a long-running job's re-attach handle survives the 101st short job.

Combined session-cookie-or-API-key auth that does not hold a DB session across the request (fixes a SQLite lock bug under the obvious get_current_active_user choice).

266 v2 backend tests + 31 translator tests pass.

Frontend

@ag-ui/client@0.0.53 pinned. controllers/API/agui/ ships the run service (run-agent.ts), the React hook (use-run-flow.ts), the canvas-state reducer (state.ts), the chat reducer (chat.ts), and the bridge (run-flow-bridge.ts) that flowStore.buildFlow drives every run through.

buildWorkflowRunRequest builds the native body. createWorkflowAgent({ body }) patches HttpAgent.requestInit on the instance so the wire body is the native shape instead of the AG-UI RunAgentInput. The RunAgentInput passed to agent.run() stays local for the client-side subscriber correlation pipeline and never reaches the network.

The bridge folds AG-UI events into the same flow-store methods the v1 path used. RUN_FINISHED and RUN_ERROR tear down the subscription on the terminal event so a server-side keepalive after the run doesn't leave the canvas stuck on isBuilding=true.

Side-channel CustomEvent (langflow.event) carries the original v1 message payloads alongside the AG-UI translation, so the playground chat-view keeps consuming its familiar v1 shape. A follow-up rewrites chat-view onto AG-UI TEXT_MESSAGE_* directly and retires the side-channel.

50 frontend jest tests cover the builder, agent factory, wire-body capture, JSON-Patch state-delta parsing, the terminal-event contract, the e2e bridge through real stores, and useRunFlow's concurrency lifecycle.

CI

CI was green on the last run before merging the latest release-1.10.0 (98 jobs, 0 failures: backend matrix on Python 3.10 + 3.14 across LFX, CLI, integration, and unit groups 1 to 5, plus the 70-shard Playwright matrix). It's re-running now after that merge.

Follow-ups

• delete api/v1/chat.py::build_flow + the v1 build pipeline
• drop customBuildUtils.ts and the eventDelivery config the canvas no longer consults
• retire the langflow.event side-channel by porting chat-view to AG-UI TEXT_MESSAGE_* directly
• inline withEventDeliveryModes at each Playwright call site and delete the shim
• document POST /api/v2/workflows

Integrating from TypeScript / JavaScript

Plain fetch examples (Node 18+ and browser) for the three execution modes, matching the real v2 wire shapes.

As a frontend/Node developer, I want to run any flow through one endpoint with sync, streaming, and background modes, so that I can pick the right execution shape per feature without learning a different API for each. The run should give a deterministic primary answer plus the full per-component map, stream tokens as they happen, let a long job disconnect/re-attach without losing events, read a stop as a cancellation (not a failure), and round-trip session_id for chat continuity.

Setup

const LANGFLOW_URL = "https://your-langflow-host";
const API_KEY = process.env.LANGFLOW_API_KEY!;
const FLOW_ID = "67ccd2be-17f0-8190-81ff-3bb2cf6508e6";

type WorkflowMode = "sync" | "stream" | "background";
type JobStatus = "queued" | "in_progress" | "completed" | "failed" | "cancelled" | "timed_out";

interface WorkflowRunRequest {
  flow_id: string;
  input_value?: string;
  mode?: WorkflowMode;
  stream_protocol?: "langflow" | "agui";
  session_id?: string;                                  // continue a chat/memory thread
  tweaks?: Record<string, Record<string, unknown>>;     // per-component param overrides
  globals?: Record<string, string>;                     // request-level globals (sync only)
  output_ids?: string[];                                // pin which output is the "answer" (sync)
}

interface WorkflowOutput { reason: "single" | "multiple" | "none" | "non_string" | "failed"; text: string | null; source: string | null; }
interface ComponentOutput { type: string; status: JobStatus; display_name: string | null; content: unknown; metadata: Record<string, unknown> | null; }
interface WorkflowExecutionResponse { flow_id: string; session_id: string | null; job_id: string | null; status: JobStatus; output: WorkflowOutput; outputs: Record<string, ComponentOutput>; errors: { error: string; code?: string }[]; has_errors: boolean; }
interface WorkflowJobResponse { job_id: string; flow_id: string; status: JobStatus; links: { status: string; events: string; stop: string }; }

Sync — one-shot answer

async function runSync(req: WorkflowRunRequest): Promise<WorkflowExecutionResponse> {
  const res = await fetch(`${LANGFLOW_URL}/api/v2/workflows`, {
    method: "POST",
    headers: { "Content-Type": "application/json", "x-api-key": API_KEY },
    body: JSON.stringify({ mode: "sync", ...req }),
  });
  if (!res.ok) throw new Error(`Workflow failed: ${res.status} ${await res.text()}`);
  return res.json();
}

const result = await runSync({ flow_id: FLOW_ID, input_value: "Summarize today's standup." });

if (result.output.reason === "single") {
  console.log(result.output.text);          // the answer
} else {
  // "multiple" | "none" | "non_string" | "failed" — output.text is null and reason says why.
  console.log(result.output.reason, result.outputs);
}

Pin a specific output so output.text is deterministic on multi-output flows: output_ids: ["ChatOutput-final"].

Streaming (AG-UI) — live tokens + tool activity

stream returns SSE. EventSource only does GET, so for a POST stream use fetch + a small SSE reader:

async function* sse(res: Response): AsyncGenerator<{ id?: string; data: any }> {
  const reader = res.body!.getReader();
  const decoder = new TextDecoder();
  let buffer = "";
  for (;;) {
    const { value, done } = await reader.read();
    if (done) break;
    buffer += decoder.decode(value, { stream: true });
    let i: number;
    while ((i = buffer.indexOf("\n\n")) !== -1) {
      const frame = buffer.slice(0, i);
      buffer = buffer.slice(i + 2);
      let id: string | undefined;
      const data: string[] = [];
      for (const line of frame.split("\n")) {
        if (line.startsWith("id:")) id = line.slice(3).trim();
        else if (line.startsWith("data:")) data.push(line.slice(5).trim());
      }
      if (data.length) yield { id, data: JSON.parse(data.join("\n")) };
    }
  }
}

async function runStream(req: WorkflowRunRequest, onToken: (t: string) => void) {
  const res = await fetch(`${LANGFLOW_URL}/api/v2/workflows`, {
    method: "POST",
    headers: { "Content-Type": "application/json", "x-api-key": API_KEY, Accept: "text/event-stream" },
    body: JSON.stringify({ mode: "stream", stream_protocol: "agui", ...req }),
  });
  if (!res.ok) throw new Error(`Stream failed: ${res.status}`);

  for await (const { data } of sse(res)) {
    switch (data.type) {                                   // AG-UI event types
      case "TEXT_MESSAGE_CONTENT": onToken(data.delta); break;
      case "TOOL_CALL_START":      console.log(`↳ tool: ${data.toolCallName}`); break;
      case "TOOL_CALL_RESULT":     console.log(`↳ result: ${data.content}`); break;
      case "CUSTOM":
        if (data.name === "langflow.run.cancelled") console.log("(run was stopped)");
        break;
      case "RUN_FINISHED":         return;                  // clean end
      case "RUN_ERROR":            throw new Error(data.message);
    }
  }
}

Prefer raw v1-shaped frames? Send stream_protocol: "langflow" and read data.event / data.data.

Background — long job, resumable

async function startBackground(req: WorkflowRunRequest): Promise<WorkflowJobResponse> {
  const res = await fetch(`${LANGFLOW_URL}/api/v2/workflows`, {
    method: "POST",
    headers: { "Content-Type": "application/json", "x-api-key": API_KEY },
    body: JSON.stringify({ mode: "background", ...req }),
  });
  return res.json();   // { job_id, status, links: { status, events, stop } }
}

// (A) Poll for the final result
async function waitForJob(job: WorkflowJobResponse): Promise<WorkflowExecutionResponse> {
  for (;;) {
    const res = await fetch(`${LANGFLOW_URL}${job.links.status}`, { headers: { "x-api-key": API_KEY } });
    const body: WorkflowExecutionResponse = await res.json();
    if (["completed", "failed", "cancelled", "timed_out"].includes(body.status)) return body;
    await new Promise((r) => setTimeout(r, 1000));
  }
}

const job = await startBackground({ flow_id: FLOW_ID, input_value: "Run the nightly report", session_id: "report-thread" });
const final = await waitForJob(job);
console.log(final.output.text, "→ continue with session:", final.session_id);

// (B) Re-attach to the event stream, resumable across disconnects
async function tail(job: WorkflowJobResponse, lastEventId?: string): Promise<string | undefined> {
  const headers: Record<string, string> = { "x-api-key": API_KEY, Accept: "text/event-stream" };
  if (lastEventId) headers["Last-Event-ID"] = lastEventId;     // replays from where you left off, no gap
  const res = await fetch(`${LANGFLOW_URL}${job.links.events}`, { headers });
  let lastId = lastEventId;
  for await (const { id, data } of sse(res)) {
    lastId = id ?? lastId;
    if (data.type === "TEXT_MESSAGE_CONTENT") process.stdout.write(data.delta);
    if (data.type === "RUN_FINISHED") break;
  }
  return lastId;  // pass back into tail() to resume if the socket dropped
}

Stopping a run

async function stopJob(job: WorkflowJobResponse) {
  await fetch(`${LANGFLOW_URL}${job.links.stop}`, {
    method: "POST",
    headers: { "Content-Type": "application/json", "x-api-key": API_KEY },
    body: JSON.stringify({ job_id: job.job_id }),
  });
}

A stopped run's replayed stream ends in CUSTOM langflow.run.cancelled + RUN_FINISHED (on the langflow protocol, a cancelled event), not RUN_ERROR, so the tail() loop treats a deliberate stop as a clean end.

Notes

• Memory: reuse the same session_id across calls; it's echoed back on every response (including completed background jobs).
• Per-run overrides: tweaks: { "OpenAIModel-x": { "temperature": 0.2 } } overrides component params without editing the flow.
• Globals: honored in sync mode; ignored for stream / background.
• Auth: examples use x-api-key; a session Authorization: Bearer <token> works too.

Endpoints

Method	Path	Purpose
POST	/api/v2/workflows	Run a flow (mode: sync / stream / background)
GET	/api/v2/workflows?job_id={id}	Background job status (returns the execution response once terminal)
GET	/api/v2/workflows/{job_id}/events	Re-attach to a background run's SSE stream (Last-Event-ID to resume)
POST	/api/v2/workflows/stop	Stop a background run ({ job_id })
POST	/api/v2/workflows/public	Run a PUBLIC flow as an unauthenticated visitor (stream-only, throttled)

[coderabbitai]

Important

Review skipped

Auto incremental reviews are disabled on this repository.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

Run ID: 258f2592-789a-441d-884a-91d2c117b23d

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

Walkthrough

This PR implements a comprehensive v2 workflow execution system using AG-UI streaming protocol. It replaces the legacy developer-API/API-key-only architecture with session-based authentication, multi-mode execution (sync/stream/background), and streaming SSE frames through pluggable adapters. The backend introduces stream adapters for protocol abstraction, AG-UI translator for event lifecycle, and background buffering with reattach support. The frontend shifts to native AG-UI event handlers, pure state reducers, and orchestrated flow bridge integration. Extensive tests validate adapter contracts, event sequences, endpoint behavior, IDOR enforcement, and end-to-end workflows.

Changes

AG-UI V2 Workflow System

Layer / File(s)	Summary
Documentation and Configuration Updates .cursor/rules/*, docs/agents/*, .github/workflows/*, src/backend/base/pyproject.toml, src/frontend/package.json	Removed legacy cursor rules; introduced new docs/agents/ guidance (philosophy, architecture, components, contracts, testing, anti-patterns); updated GitHub Actions for improved uv tree version detection and bundle building; added AG-UI protocol and client dependencies.
LFX Schema: Workflow Request Model and Validation src/lfx/src/lfx/schema/workflow.py, src/lfx/tests/unit/schema/test_workflow_run_request.py, src/lfx/src/lfx/services/settings/base.py	Introduced WorkflowMode enum and WorkflowRunRequest BaseModel for v2 execution parameters; added UUID validation and JSON schema examples; normalized CORS wildcard handling for Python 3.14 compatibility.
Backend Converters: Request Parsing and Response Building src/backend/base/langflow/api/v2/converters.py	Introduced ParsedWorkflowRun frozen dataclass for native v2 parsing; refactored run_response_to_workflow_response and create_error_response to accept raw inputs dict instead of request wrapper; removed flat-input parser.
Stream Adapter Infrastructure: Protocol Abstraction and Registry src/backend/base/langflow/api/v2/adapters/__init__.py, src/backend/base/langflow/api/v2/adapters/agui.py, src/backend/base/langflow/api/v2/adapters/langflow.py	Defined stream adapter protocol for SSE event framing with per-run context; implemented registry with registration and lookup; added AG-UI adapter wrapping translator output and Langflow adapter for v1 passthrough.
AG-UI Translator: Event Lifecycle Management src/backend/base/langflow/api/v2/agui_translator.py	Implemented stateful translator converting Langflow events to AG-UI protocol; handles run lifecycle, streamed text messages, node state snapshots via JSON-Patch deltas, tool-call lifecycle events, and custom Langflow content events with deduplication.
Authentication and Request Parsing Support src/backend/base/langflow/services/auth/utils.py	Added get_current_user_for_workflow helper for session/OAuth2 or API-key fallback without holding DB session across request; updated converters to use ParsedWorkflowRun for native v2 parsing.
V2 Workflow Endpoints: Sync/Stream/Background Execution src/backend/base/langflow/api/v2/workflow.py, src/backend/base/langflow/api/v2/workflow_reconstruction.py	Refactored POST /workflows to route by mode into sync (with timeout), stream (SSE-framed via adapter), and background (in-memory buffer with reattach); updated GET /workflows and POST /workflows/stop to use session auth; added GET /workflows/{job_id}/events for background reattach with Last-Event-ID skipping.
Backend Unit Tests: Adapters, Translator, Converters src/backend/tests/unit/api/v2/adapters/test_*.py, src/backend/tests/unit/api/v2/test_agui_translator.py, src/backend/tests/unit/api/v2/test_converters.py	Comprehensive unit tests validating adapter contracts, translator event sequences and state invariants, and v2 converter native request parsing and response building.
Backend Integration Tests: Workflow Endpoints and IDOR src/backend/tests/unit/api/v2/test_workflow.py, src/backend/tests/unit/api/v2/test_workflow_agui.py	Extensive integration tests for workflow status, stop, IDOR enforcement, AG-UI request validation, execution mode routing, streaming/sync/background behavior, cancellation, reattach with concurrent readers, job finalization detection, and background buffer eviction/clearing.
Frontend Event Handling: Chat, State, and HTTP Integration src/frontend/src/controllers/API/agui/chat.ts, src/frontend/src/controllers/API/agui/state.ts, src/frontend/src/controllers/API/agui/run-agent.ts, src/frontend/src/controllers/API/agui/run-flow-bridge.ts	Implemented pure reducers for AG-UI chat thread and canvas state; created HTTP agent wrapper and request builder for v2 workflows endpoint; wired event dispatcher for terminal and non-terminal AG-UI events and state delta application.
Frontend Lifecycle: Flow Bridge and Run Hook src/frontend/src/controllers/API/agui/use-run-flow.ts	Implemented runFlowAGUI flow bridge to orchestrate workflow runs through HTTP agent, apply state deltas, and handle terminal events; added useRunFlow React hook for component-level execution with abort support.
Frontend Unit Tests: Event Handlers and Reducers src/frontend/src/controllers/API/agui/__tests__/*.test.ts	Unit tests for chat reducer message lifecycle, canvas state delta application, HTTP agent request building with field mapping, and event handler terminal/non-terminal dispatch behavior.
Frontend Integration: Store Migration and End-to-End Tests src/frontend/src/stores/flowStore.ts, src/frontend/src/stores/__tests__/flowStore.test.ts, src/frontend/tests/utils/withEventDeliveryModes.ts	Updated flowStore to use v2 runFlowAGUI endpoint for build execution; migrated test infrastructure to support AG-UI event flow; added end-to-end tests verifying complete run sequences, SSE decoding, store updates, and failure handling.

🎯 4 (Complex) | ⏱️ ~60 minutes

Possibly Related PRs

• langflow-ai/langflow#13198: Updates same .github/workflows/nightly_build.yml by gating nightly releases on create-nightly-tag success.
• langflow-ai/langflow#13199: Includes same .github/workflows/python_test.yml timeout bump from 30 to 40 minutes.
• langflow-ai/langflow#13141: Affects same src/lfx/src/lfx/services/settings/base.py for settings refactoring and validator changes.

Suggested Reviewers

• jordanrfrazier
• dkaushik94

🚥 Pre-merge checks | ✅ 7 | ❌ 2

❌ Failed checks (2 warnings)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 61.18% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.
Test Quality And Coverage	⚠️ Warning	Tests cover main features with async patterns, but lack error-response testing (HTTP non-200 in agent), promise-settlement tests (abort), malformed-payload tests, and have fragile assertions.	Add HTTP error, abort-settlement, malformed-payload, and SSE assertion hardening tests.

✅ Passed checks (7 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title accurately describes the main change: introducing a native v2 workflows endpoint with pluggable stream protocol support.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Test Coverage For New Implementations	✅ Passed	PR includes ~5300 lines of new tests across backend (adapters, translator, workflows, schema) and frontend with proper naming conventions and substantive test coverage.
Test File Naming And Structure	✅ Passed	Backend tests use test_*.py (261 tests), frontend tests use *.test.ts (50 tests). Tests are well-organized with descriptive names, proper setup/teardown, and comprehensive error coverage.
Excessive Mock Usage Warning	✅ Passed	Core logic tests have zero mocks (1053 lines); integration tests use real database/HTTP; mocks appropriately scoped for Graph setup and thin endpoints; no test obscures actual behavior.
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feat/v2-workflows-agui

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[codecov]

Codecov Report

❌ Patch coverage is 89.38907% with 33 lines in your changes missing coverage. Please review.
✅ Project coverage is 50.41%. Comparing base (5620ce1) to head (cca42a9).
⚠️ Report is 8 commits behind head on main.

Files with missing lines	Patch %	Lines
src/backend/base/langflow/api/v2/workflow.py	89.33%	16 Missing ⚠️
...rc/backend/base/langflow/api/v2/agui_translator.py	89.51%	13 Missing ⚠️
src/backend/base/langflow/services/auth/utils.py	66.66%	4 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff             @@
##             main   #13307       +/-   ##
===========================================
- Coverage   53.12%   50.41%    -2.72%     
===========================================
  Files        2033      619     -1414     
  Lines      184171    58166   -126005     
  Branches    26195     5052    -21143     
===========================================
- Hits        97843    29324    -68519     
+ Misses      85219    27732    -57487     
- Partials     1109     1110        +1     

Flag	Coverage Δ
backend	50.76% <89.38%> (-5.46%)	⬇️
frontend	?
lfx	50.06% <ø> (-0.01%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
src/backend/base/langflow/api/v2/converters.py	87.71% <100.00%> (-5.44%)	⬇️
...nd/base/langflow/api/v2/workflow_reconstruction.py	84.00% <ø> (-8.60%)	⬇️
src/backend/base/langflow/services/auth/utils.py	85.79% <66.66%> (-5.35%)	⬇️
...rc/backend/base/langflow/api/v2/agui_translator.py	89.51% <89.51%> (ø)
src/backend/base/langflow/api/v2/workflow.py	75.35% <89.33%> (-9.62%)	⬇️

... and 1543 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[codecov]

Codecov Report

❌ Patch coverage is 94.64706% with 91 lines in your changes missing coverage. Please review.
⚠️ Please upload report for BASE (release-1.11.0@75e0115). Learn more about missing BASE report.

Files with missing lines	Patch %	Lines
...backend/base/langflow/api/v2/workflow_execution.py	88.71%	22 Missing ⚠️
...ontend/src/controllers/API/agui/run-flow-bridge.ts	96.48%	17 Missing ⚠️
src/backend/base/langflow/api/v2/workflow.py	83.11%	13 Missing ⚠️
...ackend/base/langflow/api/v2/workflow_background.py	93.71%	11 Missing ⚠️
...rc/backend/base/langflow/api/v2/workflow_public.py	89.65%	6 Missing ⚠️
...rc/backend/base/langflow/api/v2/agui_translator.py	97.42%	5 Missing ⚠️
src/backend/base/langflow/services/auth/utils.py	58.33%	5 Missing ⚠️
src/backend/base/langflow/api/v2/converters.py	96.72%	2 Missing ⚠️
...ackend/base/langflow/api/v2/workflow_validation.py	95.55%	2 Missing ⚠️
...c/lfx/src/lfx/services/settings/groups/security.py	50.00%	1 Missing and 1 partial ⚠️
... and 6 more

Additional details and impacted files

@@                Coverage Diff                @@
##             release-1.11.0   #13307   +/-   ##
=================================================
  Coverage                  ?   58.49%           
=================================================
  Files                     ?     2319           
  Lines                     ?   221664           
  Branches                  ?    32867           
=================================================
  Hits                      ?   129654           
  Misses                    ?    90539           
  Partials                  ?     1471           

Flag	Coverage Δ
backend	66.45% <92.54%> (?)
frontend	57.34% <97.56%> (?)
lfx	54.66% <75.00%> (?)

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
src/backend/base/langflow/api/router.py	100.00% <100.00%> (ø)
src/backend/base/langflow/api/v2/adapters/agui.py	100.00% <100.00%> (ø)
src/frontend/src/controllers/API/agui/run-agent.ts	100.00% <100.00%> (ø)
...c/lfx/src/lfx/custom/custom_component/component.py	59.46% <100.00%> (ø)
...rc/lfx/src/lfx/services/settings/groups/runtime.py	100.00% <100.00%> (ø)
src/backend/base/langflow/api/build.py	85.27% <92.85%> (ø)
src/backend/base/langflow/api/utils/flow_utils.py	80.15% <95.00%> (ø)
src/backend/base/langflow/api/v1/chat.py	51.58% <0.00%> (ø)
.../backend/base/langflow/api/v2/adapters/langflow.py	97.95% <97.95%> (ø)
...nd/base/langflow/api/v2/workflow_reconstruction.py	95.45% <95.00%> (ø)
... and 11 more

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[Copilot]

Pull request overview

This PR introduces an end-to-end AG-UI execution path for Langflow workflows by routing the frontend “run/build” flow through the new POST /api/v2/workflows endpoint, which can stream AG-UI events over SSE (and also supports sync/background modes). It adds a backend event translator and supporting API changes, plus a frontend bridge/reducers/tests to consume AG-UI while keeping the existing v1 build path intact when the feature flag is off.

Changes:

• Backend: Add AG-UI RunAgentInput contract handling for POST /api/v2/workflows, including streaming SSE translation, background buffering + reattach, and updated auth dependency to avoid long-lived DB sessions.
• Frontend: Add @ag-ui/client integration (agent + bridge + reducers/hooks) and gate the new run path behind LANGFLOW_V2_WORKFLOWS_AGUI_ENABLED.
• Tests: Add/reshape unit + integration coverage for the translator, request contract, and parsing logic; adjust existing v2 workflow tests accordingly.

Reviewed changes

Copilot reviewed 25 out of 27 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
uv.lock	Adds Python dependency resolution entries for ag-ui-protocol.
src/backend/base/pyproject.toml	Pins ag-ui-protocol==0.1.18 for backend AG-UI types/events.
src/backend/base/langflow/services/auth/utils.py	Adds get_current_user_for_workflow (session-or-API-key auth without holding a DB session).
src/backend/base/langflow/api/v2/workflow.py	Implements AG-UI RunAgentInput handling, SSE streaming translation, background buffering + /events reattach, and updates execution flow.
src/backend/base/langflow/api/v2/converters.py	Adds ParsedWorkflowRun and parse_run_agent_input; updates converters to accept inputs dict directly.
src/backend/base/langflow/api/v2/agui_translator.py	New translator mapping Langflow EventManager events to AG-UI protocol events.
src/backend/base/langflow/api/v2/workflow_reconstruction.py	Updates reconstruction to use new converter signature (inputs={}) instead of WorkflowExecutionRequest.
src/backend/tests/unit/api/v2/test_workflow.py	Reduces/refocuses v2 tests to status/stop/IDOR; defers AG-UI contract tests to dedicated module.
src/backend/tests/unit/api/v2/test_workflow_agui.py	New no-mocks endpoint tests for AG-UI request contract + mode dispatch + streaming/background behaviors.
src/backend/tests/unit/api/v2/test_run_agent_input.py	New unit tests for parse_run_agent_input.
src/backend/tests/unit/api/v2/test_converters.py	Updates converter tests for new function signatures.
src/backend/tests/unit/api/v2/test_agui_translator.py	New unit tests for AG-UI translator correctness and well-formed event streams.
src/frontend/package.json	Pins @ag-ui/client@0.0.53.
src/frontend/package-lock.json	Locks @ag-ui/* dependency tree for the frontend.
src/frontend/vite.config.mts	Exposes LANGFLOW_V2_WORKFLOWS_AGUI_ENABLED to the frontend build.
src/frontend/src/customization/feature-flags.ts	Adds ENABLE_V2_WORKFLOWS_AGUI feature flag.
src/frontend/src/stores/flowStore.ts	Routes buildFlow through runFlowAGUI when the flag is enabled.
src/frontend/src/controllers/API/agui/run-agent.ts	Adds AG-UI workflow agent wrapper + buildRunInput.
src/frontend/src/controllers/API/agui/run-flow-bridge.ts	Adds bridge to fold AG-UI events into existing flowStore methods.
src/frontend/src/controllers/API/agui/state.ts	Adds pure reducer for AG-UI STATE_SNAPSHOT/STATE_DELTA canvas state.
src/frontend/src/controllers/API/agui/chat.ts	Adds pure reducer for AG-UI TEXT_MESSAGE_* chat lifecycle events.
src/frontend/src/controllers/API/agui/use-run-flow.ts	Adds a React hook around the AG-UI agent run/abort lifecycle.
src/frontend/src/controllers/API/agui/tests/state.test.ts	Adds tests for AG-UI canvas-state reducer.
src/frontend/src/controllers/API/agui/tests/chat.test.ts	Adds tests for AG-UI chat reducer.
src/frontend/src/controllers/API/agui/tests/run-agent.test.ts	Adds tests for buildRunInput and agent construction.
src/frontend/tests/utils/withEventDeliveryModes.ts	Collapses the v1 event-delivery-mode matrix to a single run when AG-UI is enabled.
.secrets.baseline	Updates baseline metadata/line number due to file changes.

Files not reviewed (1)

• src/frontend/package-lock.json: Language not supported

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[dkaushik94]

Suggestion: introduce an event-sourcing surface in EventManager for protocol adapters

Problem

Langflow's EventManager was designed for state replication: components emit add_message snapshots of a persisted Message, and the v1 frontend mirrors DB state from those snapshots. Fit-for-purpose for v1, but it forces every protocol adapter (AG-UI today; OpenAI Responses, Anthropic, MCP tomorrow) to reconstruct an explicit message lifecycle by inference:

• "first token of an id" → infer message_start
• "different id arrives" → infer previous message_end
• "add_message with state=complete" → infer finalize
• plus dedupe against re-fires of the same id

The reconstruction is leaky. AGUITranslator's singular _open_message_id silently drops tokens and finalizer text when two messages stream in parallel — see agui_translator.py#L111-L135. The same reconstruction tax will be paid independently by every future adapter.

Suggestion

Add a parallel, opt-in event vocabulary alongside the existing one:

Event	Payload
message.start	{ id, role, parent_message_id? }
message.delta	{ id, chunk }
message.end	{ id, final_text?, usage? }
tool.start	{ tool_call_id, message_id, name, args }
tool.end	{ tool_call_id, result | error }
content.block	{ id, type, data } for json / code / media

• Components opt in via a new Component.stream_message_v2 helper.
• Legacy send_message / token / add_message keep working untouched — v1 frontend is unaffected.
• Protocol adapters consume the explicit lifecycle 1:1 — no inference, no dedupe, no singular-open invariant — and naturally support parallel streams via id multiplexing.

This is the industry-standard pattern for multi-stream protocols:

• Anthropic Messages API — content-block index
• OpenAI Responses API — item_id / output_index
• AG-UI — message_id, tool_call_id
• LangChain astream_events — run_id, parent_ids

Why now

• Cost is bounded: two event paths in EventManager during a transition period.
• Value compounds with every new adapter (AG-UI, OpenAI Responses, MCP, …).
• Unblocks parallel-agent flows on AG-UI without protocol violations.

Tactical bridge (this PR scope, optional)

Switch AGUITranslator from a singular _open_message_id: str | None to _open_message_ids: set[str] — preserves the dual-emission dedupe (the real LF-specific reason _emitted_text_message_ids exists) while removing the close-on-id-switch heuristic that causes the drops. ~10-line change; fixes the immediate AG-UI bug without waiting on the larger event-vocabulary work.

Out of scope

• Migrating persistence to event-sourcing wholesale (DB write becomes a consumer of the stream). Right model in the abstract; breaks v1 and every existing component. Not proposed here.

[ogabrielluiz]

Hey @dkaushik94, thanks for the really thoughtful write-up. Two parts here, and I want to answer both.

On the tactical bridge: the drop you spotted is already fixed on the branch, and it went a bit further than the set[str] you suggested. The translator now buffers parallel message ids in an ordered _buffered_messages dict and promotes/drains them when the open message genuinely ends, instead of just tracking multiple open ids (agui_translator.py, _translate_token / _drain_messages / _promote_next_buffered). The reason it isn't just a set is the finalizer text problem: a parallel message can complete via add_message while it's still waiting for the wire, so we hold its authoritative final_text and emit the full START/CONTENT/END trio on promotion rather than reconstructing it. remove_message tombstones a buffered id too, so a retracted partial doesn't flush later. There's coverage for the exact interleaving case you described in test_agui_translator.py (test_parallel_interleaved_tokens_preserve_all_content, test_token_for_second_message_is_buffered_until_first_closes, test_end_drains_buffered_messages_before_run_finished). I think your comment was against the older singular _open_message_id code, this landed a few hours later the same day.

On the bigger idea: I think you're right that the inference tax is real and it compounds per adapter. An explicit message.start/delta/end + tool.start/end + content.block vocabulary that adapters consume 1:1 is the right shape, and keeping it opt-in alongside send_message/token/add_message so v1 is untouched is the part that makes it actually shippable. I don't want to fold it into this PR since the buffering fix already unblocks parallel streams on AG-UI without protocol violations, but I'd like to do it as its own piece of work. Could you file it as a separate issue (or I can), with the Component.stream_message_v2 helper and the EventManager dual-path as the surface? That keeps this PR scoped and gives the larger change room to be reviewed on its own.

[Cristhianzl]

⛔ Blockers (resolve before merge)

B1 — workflow.py exceeds file-structure hard limits (LOC, classes, mixed responsibilities)

File: src/backend/base/langflow/api/v2/workflow.py (whole file, ~1,565 lines / ~1,270 code lines)
Issue: The PR grows this module by +977 lines to ~1,270 code lines. It holds 26 top-level symbols including two classes (_WorkflowEventQueue, _BackgroundRun) and four clearly different responsibility groups in one file: HTTP route handlers (execute_workflow, get_workflow_status, stop_workflow, reattach_workflow_events), request validation (_validate_flow_data_for_execution, _validate_output_ids, _reject_unsupported_sync_fields, _enforce_flow_data_override_owner), execution orchestration (execute_sync_workflow*, _stream_event_frames, execute_workflow_background), and an in-memory background-run buffer + registry (_BackgroundRun, _register_background_run, _finalize_job_status, _clear_background_run, _finish_cancelled_background_run).
Why it matters: The repo hard rules cap a code file at 500 LOC (600–700 only when SRP holds), 1 main class, and forbid mixing validate* with execution/persistence prefixes in one file. At 1,270 LOC with two stateful classes and four responsibility groups, none of those hold. This is the highest-traffic module in the PR (a new public endpoint + streaming), so the maintenance and review cost compounds. The hook that enforces the 500-LOC cap blocks at >700.
Suggested fix: Split by responsibility, e.g.:

• workflow.py — route handlers only (the four @router functions), delegating to the modules below.
• workflow_execution.py — execute_sync_workflow*, _build_run_inputs, _resolve_request_variables, _stream_event_frames, _execute_streaming_workflow.
• workflow_background.py — _BackgroundRun, _WorkflowEventQueue, registry helpers (_register_background_run, _finalize_job_status, _clear_background_run, _finish_cancelled_background_run, _buffer_background_run).
• workflow_validation.py — the _validate_* / _reject_* / _enforce_* guards.

Top-level symbols (grep)

class _WorkflowEventQueue:        (647)
async def _stream_event_frames(   (694)
class _BackgroundRun:             (880)
async def execute_workflow(       (252)  async def get_workflow_status( (1219)
async def stop_workflow(          (1378) async def reattach_workflow_events( (1494)
def _validate_flow_data_for_execution / _validate_output_ids / _reject_unsupported_sync_fields / _enforce_flow_data_override_owner
def _register_background_run / _finalize_job_status / _clear_background_run / _finish_cancelled_background_run

---

⚠️ Important (preferably this PR)

I1 — No execution timeout on stream / background / public modes (unbounded LLM runtime dependency)

File: src/backend/base/langflow/api/v2/workflow.py:319-360, :1067-1147; src/backend/base/langflow/api/v2/workflow_public.py:173-188
Issue: Only sync mode is wrapped in asyncio.wait_for(..., timeout=EXECUTION_TIMEOUT). stream, background, and the entire public endpoint drive generate_flow_events (which executes the graph, including any LLM components) with no upper bound on run time.
Why it matters: A flow that calls an LLM is a network call to a slow, non-deterministic external service. On the public endpoint an anonymous visitor can start a run that holds a server task (and a process-local buffer for background) indefinitely. With no per-run wall-clock ceiling, a hung upstream provider or a deliberately slow consumer ties up resources. The repo's AI-runtime-resilience rules require an explicit timeout on every path that runs an LLM, not just sync.
Suggested fix: Apply a wall-clock ceiling to the stream/background/public drive loop too (e.g. cancel run_task after a configurable STREAM_EXECUTION_TIMEOUT, emitting the adapter's terminal-error event). At minimum document why streaming is intentionally unbounded if that is the decision, and gate the public endpoint with a tighter cap.

I2 — Public streaming endpoint has no rate limiting and no per-run cost ceiling

File: src/backend/base/langflow/api/v2/workflow_public.py:65-188
Issue: POST /api/v2/workflows/public is anonymous (owner-impersonated) and, combined with I1, has no rate limit, no concurrency cap per client/IP, and no token/cost ceiling. The app wires slowapi globally (main.py) but no @limiter.limit is applied here.
Why it matters: This is a new unauthenticated, externally-reachable endpoint that executes flows under the owner's credentials and can invoke paid LLM calls. Without a rate limit it is an amplification and cost-exhaustion vector against the flow owner's provider keys. The security checklist lists "rate limiting on all public-facing endpoints" as a required control.
Suggested fix: Confirm whether v1 build_public_tmp carries a limit and reach at least parity; if neither has one, add a @limiter.limit(...) to the public endpoint (per client_id / IP) and a per-run cost/iteration ceiling. If the decision is to rely on an upstream gateway, capture that in the PR description.

I3 — Internal exception messages leaked to clients in error bodies

File: src/backend/base/langflow/api/v2/workflow.py:427-436, :375-384, :1256-1263, :1362-1370, :1477-1485
Issue: Several catch-all handlers embed str(exc) / f"{err!s}" / f"{e!s}" directly in the response message (e.g. "An unexpected error occurred: {err!s}", "Failed to retrieve job from database: {exc!s}", "Failed to stop job: {job_id} - {exc!s}").
Why it matters: "No internal details exposed in error messages to end users" is a general security control. Raw exception text from the DB layer or graph build can leak table names, file paths, driver internals, or stack-derived detail to an API client (and, via the streamed error event, to anonymous public-flow visitors). Note the public endpoint deliberately sanitizes the blocked-component message to "This flow cannot be executed." — the authenticated handlers are inconsistent with that posture.
Suggested fix: Log the full exc server-side (already partly done) and return a generic, code-tagged message to the client ("Internal server error" + the existing code), keeping the detailed string out of the body. Apply the same to the langflow/agui adapter error_events(str(error)) path that reaches the wire.

I4 — _resolve_request_variables / body globals reach component runtime without validation beyond length

File: src/backend/base/langflow/api/v2/workflow.py:230-241, :532-535; src/lfx/src/lfx/schema/workflow.py:245-254
Issue: Body globals (and X-LANGFLOW-GLOBAL-VAR-* headers) are merged and injected into graph.context["request_variables"] for every authenticated run. The only constraint is key/value length (GLOBAL_KEY_MAX_LEN / GLOBAL_VALUE_MAX_LEN). The header path in extract_global_variables_from_headers is not visible in this diff and is trusted as-is.
Why it matters: These values become component inputs at runtime (the trust-boundary question: "who controls each value, and could they lie?"). Caller-controlled globals that silently override a flow's configured global variables is a privilege/data-exposure surface if any component treats a global as trusted (e.g. a connection string or a path). The PR description and code don't state which globals are allowed to be overridden per-request.
Suggested fix: Document and, if needed, allowlist which global keys a request may set, or confirm in a Why: comment / PR ## Design Decisions that request-level globals are intentionally unrestricted and that no component treats a global as a trust boundary. This is the kind of assumption the comprehension audit asks to be captured durably.

---

💡 Recommended (can ship as a follow-up)

R1 — Misleading "commented out" comment over live dataframe code; ungated TODOs

File: src/backend/base/langflow/api/v2/converters.py:262-274, :366-367
Issue: The comment says "The following code is commented out pending further requirements analysis" immediately above a if output_type == "dataframe": block that is not commented out and actually executes. There are also two # TODO: Future scope comments with no ticket reference.
Why it matters: A comment that contradicts the code is worse than no comment — a future maintainer will trust the wrong statement. The repo rules forbid WHAT-comments and TODOs without a ticket.
Suggested fix: Delete the misleading comment, keep the executing dataframe branch (or remove it if truly unused), and either drop the TODOs or attach a tracking issue id.

R2 — Likely-dead legacy schema WorkflowExecutionRequest

File: src/lfx/src/lfx/schema/workflow.py:113-178
Issue: WorkflowExecutionRequest (with validate_execution_mode, background/stream booleans, flat inputs) is only referenced by the lazy __getattr__ export in lfx/schema/__init__.py; the new endpoint uses WorkflowRunRequest. No production caller in the diff constructs it.
Why it matters: YAGNI / dead-code: a ~65-line request model with its own validator that nothing executes invites confusion about which is the real request shape.
Suggested fix: If nothing outside tests uses it, remove it (and its export) in this PR or file a follow-up to retire it once the v1 build path is deleted (the PR already lists that as a follow-up).

R3 — agui adapter cancel_events routes cancellation through the generic error translation

File: src/backend/base/langflow/api/v2/adapters/agui.py:508-511
Issue: Cancellation emits a RUN_ERROR via translate("error", {"error": reason}). A user-initiated stop is then indistinguishable from a real failure on the wire (both become RUN_ERROR), and the buffer's terminal_error_type == "RUN_ERROR" will mark a cancelled job as errored unless the CANCELLED guard in _finalize_job_status wins the race.
Why it matters: Clients and analytics can't tell "user cancelled" from "run failed". The _finalize_job_status CANCELLED-protection comment acknowledges this is a race.
Suggested fix: Emit a distinct cancellation signal (an AG-UI CUSTOM langflow.cancelled event, or rely on the status row) so cancel and error are separable downstream; keep the RUN_ERROR only if AG-UI genuinely has no cancellation primitive and document that.

R4 — reattach_workflow_events 409 message reveals worker-affinity internals

File: src/backend/base/langflow/api/v2/workflow.py:1540-1552
Issue: The 409 body explains "Buffered events ... are not available on this worker ... route ... requests to the worker that accepted the background run."
Why it matters: Minor information disclosure about deployment topology to any caller. Low severity but inconsistent with the privacy-first posture elsewhere (404-on-cross-user).
Suggested fix: Keep the actionable hint ("use the status endpoint") but drop the worker-routing internals from the client-facing message; log them server-side instead.

---

🟢 Nice-to-have

N1 — WorkflowStreamEvent schema appears unused by the actual stream wire shape

File: src/lfx/src/lfx/schema/workflow.py:432-493
Issue: WorkflowStreamEvent (type/run_id/timestamp/raw_event) is advertised in the OpenAPI text/event-stream schema, but the real frames are adapter-specific ({"event","data"} for langflow, AG-UI events for agui). The documented schema doesn't match either wire shape.
Suggested fix: Either align the OpenAPI example with one real protocol's frame or annotate that the SSE body shape depends on stream_protocol.

N2 — parse_workflow_run_request is a thin field-copy that duplicates the request model

File: src/backend/base/langflow/api/v2/converters.py:72-98
Issue: ParsedWorkflowRun mirrors WorkflowRunRequest field-for-field; the parse step is mostly a rename layer (mode.value, run_id=None).
Suggested fix: Acceptable as a boundary DTO, but consider whether the route could carry WorkflowRunRequest plus a resolved run_id directly to cut one mapping layer. Low priority.

[erichare]

Reviewed the full incremental diff — backend v2 endpoint, stream adapters, AG-UI translator, converters, background-job reconstruction, the auth refactor, and the frontend AG-UI bridge — plus a dedicated security pass on the public endpoint and auth changes.

Security: the public path is clean on the RCE surface. validate_public_flow_no_code_execution() (blocking PythonREPL*/PythonCodeStructuredTool/Smart Transform plus transitive RunFlow/SubFlow/FlowTool) runs before any graph build; the endpoint is gated to PUBLIC flows only; PublicWorkflowRunRequest is extra="forbid" with no data/tweaks, so a caller can't inject nodes; session IDs stay namespaced; and get_current_user_for_workflow doesn't weaken token/active-user validation. No auth bypass found.

No CRITICAL issues. The inline notes below are correctness/robustness items. Frontend terminal-event teardown, the extra="forbid" wire body, unknown-protocol -> 422 across all modes, and stream-exception -> terminal-frame handling all checked out, and the background/translator tests are meaningful rather than shallow.

[erichare]

[HIGH] Background status reconstruction always returns session_id: null. run_response_to_workflow_response echoes RunResponse.session_id into the API response, but reconstruct_workflow_response_from_job_id builds RunResponse(outputs=..., session_id=None). So GET /api/v2/workflows?job_id=... returns session_id: null for every completed background job, even though the run executed under a real session.

The schema documents this field as the handle to continue the same chat/memory thread — and background is exactly the mode where the client wasn't streaming and most needs to recover it. The session is known to the job/build rows; thread it through instead of hardcoding None.

[ogabrielluiz]

Hey @erichare, good catch. The session isn't on the job or vertex_build columns, but it's persisted inside the terminal message data, so reconstruction pulls it from there now. Data-only flows stay null since there's no thread to continue. Added an e2e test that runs a background flow and asserts the status echoes the real session.

[erichare]

[MEDIUM] Evicting a still-running background run orphans its buffer writer. The eviction here only pops the registry entry — it doesn't finish()/cancel the evicted run's _buffer_background_run coroutine, which holds a direct reference to the _BackgroundRun and keeps appending frames.

Under sustained load (>100 concurrent background jobs) the evicted buffer keeps growing with no registry entry and no reader able to find it, and a re-attach to that job now 409s even though the run is healthy — which partially defeats the bounded-memory guarantee this registry is meant to provide. When evicting a still-running entry, also cancel its queue job (or at least finish() the buffer) so the orphaned coroutine stops writing.

[ogabrielluiz]

Right, popping the entry without stopping the writer leaks. The fallback now cancels the evicted run's queue job so the buffer coroutine actually stops. Had to make _register_background_run async for it.

[erichare]

[MEDIUM] Per-component status is hardcoded COMPLETED. build_component_output sets status=JobStatus.COMPLETED for every terminal vertex; the only failure signal is the top-level error path when execute_with_status raises.

A graph that completes with a vertex that produced an error artifact without raising (a partial failure — the scenario the two-tier error handling advertises) still reports outputs[id].status == COMPLETED, so a client trusting per-component status gets false positives. Consider deriving status from the vertex valid flag, which the stream path already has.

[ogabrielluiz]

Fixed the hardcoded COMPLETED. Status comes from the error artifact / valid flag now, and the langflow adapter stops dropping valid so it lines up with what the agui translator already does. One caveat: I couldn't reproduce a non-raising partial failure reaching the sync converter on this branch (component errors raise and hit the top-level FAILED path), so this is really hardening the contract and fixing the stream/agui divergence, not an observed sync false-positive.

[erichare]

[MEDIUM] Request-body globals is honored in sync mode only. This sync path merges parsed.globals into graph context, but the stream/background paths call generate_flow_events and never pass parsed.globals anywhere — so globals is silently dropped for mode=stream/background.

The field description makes no mode distinction ("Body globals always win over the legacy headers"). Either honor globals on all paths, or document the sync-only limitation in the field description the way output_ids already documents "Ignored for stream/background".

[ogabrielluiz]

I went with documenting the sync-only limitation like output_ids does. Honoring globals on stream/background means threading context through generate_flow_events and the shared build_graph_* helpers that the v1 build endpoint also uses, so I'd rather do that as its own change. Description now says honored in sync, ignored for stream/background.

[erichare]

[MEDIUM] User-cancel is replayed as RUN_ERROR in the AG-UI stream. cancel_events routes through translate("error", ...) -> RunErrorEvent. The job row correctly lands CANCELLED, but a re-attaching client sees a stream that ends in RUN_ERROR, indistinguishable from a genuine failure — the frontend bridge then fires "Workflow run failed" and marks nodes ERROR for a deliberate stop.

AG-UI has no cancel primitive, but emitting RUN_FINISHED after closing any open messages — or a CUSTOM cancel marker — would be more honest than RUN_ERROR. The langflow adapter has the same issue.

[ogabrielluiz]

Agreed, replaying a stop as RUN_ERROR was misleading. The agui path now closes any open text, emits a CUSTOM langflow.run.cancelled marker, then RUN_FINISHED, and the langflow adapter emits a cancelled event instead of error. So a re-attaching client can tell a stop from a real failure.

[erichare]

[MEDIUM] No rate limiting + unbounded input on the unauthenticated public endpoint. This endpoint has no per-IP/per-flow rate limiting, and PublicWorkflowRunRequest.input_value/session_id have no max_length.

An anonymous caller can trigger concurrent flow executions (each running as the flow owner — real CPU/DB/LLM-credit cost) and post arbitrarily large strings that are held in memory and persisted to MessageTable. The per-run event queue is bounded (256) but the number of concurrent runs is not. This mirrors the v1 public endpoint, but v2 replicates the exposure rather than improving on it — suggest a configurable per-IP limit plus StringConstraints(max_length=...) on the public request fields, consistent with the 64 KB GlobalVarValue bound.

[ogabrielluiz]

Did both halves. input_value/session_id have max_length now (64KB / 256, matching GlobalVarValue), and there's a per-IP throttle with its own knob (public_flow_rate_limit_per_minute, default 20/min) so it doesn't borrow the login limit. It runs before any DB work.

[erichare]

[MEDIUM] reattach_workflow_events gates on the buffer owner only, bypassing the RBAC layer. When a local bg_run exists, access is gated solely by bg_run.user_id != str(current_user.id), unlike the sibling handlers in this file that route through ensure_flow_permission/share-aware fetch.

Under an authz plugin, a user holding a share on the flow (who can read its job status) can't re-attach to its event stream, and the owner check ignores RBAC entirely. Worth aligning with the other handlers, or adding a comment that stream re-attach is intentionally owner-only. (Access is correctly user-scoped, so this is a consistency/feature note, not an IDOR.)

[ogabrielluiz]

This one's deliberate so I added a comment. The live siblings (stop, active-status) are owner-only too; only the COMPLETED-status branch is share-aware because it reloads the flow, and a share-holder can still tail via the status endpoint. Making the live stream share-aware would mean touching stop and active-status to stay consistent. Do you think it's worth it, or is owner-only fine here?

[erichare]

LGTM @ogabrielluiz !

[ogabrielluiz]

Hey @Cristhianzl, thanks for the thorough review. Pushed a21d0d9 with the blockers addressed.

B1 (split): workflow.py now holds just the four route handlers (~660 lines). The validation guards moved to workflow_validation, the sync/stream run loop to workflow_execution, and the background buffer + registry to workflow_background. The layering is acyclic (validation, then execution, then background, then routes).

I1 (timeouts): added a workflow_execution_timeout setting (default 300) and applied a single wall-clock ceiling inside _stream_event_frames, so stream, background and public all inherit it. Sync already had one and now reads the same setting. A timeout comes out as the protocol's terminal error (and marks a background job failed), not a hang.

I3 (error leak): the route handlers don't put str(exc) in the response body anymore. They return a generic, code-tagged message and log the full exception server-side.

I2 and R3 landed in the earlier round on this PR: the public endpoint has a per-IP throttle with its own knob (public_flow_rate_limit_per_minute), and a stop now emits a CUSTOM langflow.run.cancelled + RUN_FINISHED instead of RUN_ERROR.

I also folded in two of the recommended ones since I was already in those files: R1 (removed the "commented out / future scope" comments sitting over live dataframe code) and R4 (the reattach 409 just points at the status endpoint now, no worker-routing detail).

A few I'd like your read on:

I4 (request globals): I added max_length on input_value/session_id and documented globals as sync-only. The values are caller-owned and nothing treats a global as a trust boundary, so I left them unrestricted instead of allowlisting. Do you think that's worth a Why: comment in the code, or is the field doc enough?

The streamed terminal error and the sync 200 body still return str(error) for component failures (the documented two-tier contract). That's the owner's own run, so I treated it as debugging detail and only sanitized the timeout case. Should the component error text be sanitized on the wire too, or is owner-facing detail fine there?

R2 (dead WorkflowExecutionRequest): it's only reachable through the lazy export. I'd rather retire it together with the v1 build-path removal that's already a follow-up, unless you'd prefer it gone in this PR.

Full v2 suite is green, plus new tests for the timeout terminal-error path and the error-body sanitization.

[dkaushik94]

LGTM. My concerns are addressed. I see others also contributed meaningfully, so I think we should be feeling great on this PR, Gabe!

[Cristhianzl]

lgtm