diff --git a/src/backend/base/langflow/api/v2/workflow.py b/src/backend/base/langflow/api/v2/workflow.py
index 827e8e7d18..4ca40dd371 100644
--- a/src/backend/base/langflow/api/v2/workflow.py
+++ b/src/backend/base/langflow/api/v2/workflow.py
@@ -51,13 +51,7 @@ from lfx.schema.workflow import (
     WorkflowStopResponse,
 )
 from lfx.services.deps import injectable_session_scope_readonly
-from pydantic_core import ValidationError as PydanticValidationError
-from sqlalchemy.exc import OperationalError
-
-from langflow.api.build import generate_flow_events
-from langflow.api.utils import extract_global_variables_from_headers
-from langflow.api.v1.schemas import FlowDataRequest, RunResponse
-from langflow.api.v2.adapters import (
+from lfx.workflow.adapters import (
     STREAM_ADAPTERS,
     StreamAdapter,
     StreamAdapterContext,
@@ -66,14 +60,26 @@ from langflow.api.v2.adapters import (
     available_protocols,
     get_stream_adapter,
 )
-from langflow.api.v2.converters import (
+from lfx.workflow.converters import (
     ParsedWorkflowRun,
     create_error_response,
     parse_workflow_run_request,
     run_response_to_workflow_response,
     workflow_response_from_output_events,
 )
+from pydantic_core import ValidationError as PydanticValidationError
+from sqlalchemy.exc import OperationalError
+
+from langflow.api.build import generate_flow_events
+from langflow.api.utils import extract_global_variables_from_headers
+from langflow.api.v1.schemas import FlowDataRequest, RunResponse
 from langflow.api.v2.workflow_reconstruction import reconstruct_workflow_response_from_job_id
+from langflow.api.v2.workflow_validation import (
+    _enforce_flow_data_override_owner,
+    _flow_not_found_privacy_exception,
+    _reject_unsupported_sync_fields,
+    _validate_flow_data_for_execution,
+)
 from langflow.exceptions.api import (
     WorkflowQueueFullError,
     WorkflowResourceError,
@@ -234,14 +240,21 @@ async def execute_workflow(
             current_user.id,
             widen_for_shares=True,
         )
-        await ensure_flow_permission(
-            current_user,
-            FlowAction.EXECUTE,
-            flow_id=flow.id,
-            flow_user_id=flow.user_id,
-            workspace_id=getattr(flow, "workspace_id", None),
-            folder_id=getattr(flow, "folder_id", None),
-        )
+        try:
+            await ensure_flow_permission(
+                current_user,
+                FlowAction.EXECUTE,
+                flow_id=flow.id,
+                flow_user_id=flow.user_id,
+                workspace_id=getattr(flow, "workspace_id", None),
+                folder_id=getattr(flow, "folder_id", None),
+            )
+        except HTTPException as exc:
+            raise _flow_not_found_privacy_exception(exc, parsed.flow_id) from exc
+
+        _reject_unsupported_sync_fields(parsed)
+        _enforce_flow_data_override_owner(parsed, flow, current_user)
+        _validate_flow_data_for_execution(parsed, flow)
 
         if parsed.mode == "sync":
             return await execute_sync_workflow_with_timeout(
diff --git a/src/backend/base/langflow/api/v2/workflow_public.py b/src/backend/base/langflow/api/v2/workflow_public.py
index 52c9686eee..af19edc464 100644
--- a/src/backend/base/langflow/api/v2/workflow_public.py
+++ b/src/backend/base/langflow/api/v2/workflow_public.py
@@ -40,20 +40,21 @@ from lfx.utils.flow_validation import (
     validate_flow_for_current_settings,
     validate_public_flow_no_code_execution,
 )
-
-from langflow.api.utils.flow_utils import (
-    scope_session_to_namespace,
-    validate_public_files,
-    verify_public_flow_and_get_user,
-)
-from langflow.api.v2.adapters import (
+from lfx.workflow.adapters import (
     STREAM_ADAPTERS,
     StreamAdapterContext,
     UnknownStreamProtocolError,
     available_protocols,
     get_stream_adapter,
 )
-from langflow.api.v2.converters import ParsedWorkflowRun
+from lfx.workflow.converters import ParsedWorkflowRun
+from limits import parse
+
+from langflow.api.utils.flow_utils import (
+    scope_session_to_namespace,
+    validate_public_files,
+    verify_public_flow_and_get_user,
+)
 from langflow.services.auth.utils import get_current_user_optional
 from langflow.services.database.models.flow.model import Flow
 from langflow.services.database.models.user.model import User, UserRead
@@ -62,6 +63,32 @@ from langflow.services.deps import get_settings_service, session_scope
 router = APIRouter(prefix="/workflows/public", tags=["Workflow (public)"])
 
 
+def _enforce_public_rate_limit(http_request: Request) -> None:
+    """Throttle anonymous public-flow runs per client IP.
+
+    Each run executes as the flow owner (real CPU/DB/LLM-credit cost), so an
+    unauthenticated caller must not be able to spin up unbounded concurrent
+    executions. Mirrors the manual limiter check the login endpoint uses, but
+    keyed to its own configurable per-minute limit under a dedicated namespace so
+    it never shares a bucket with login. No-op when rate limiting is disabled.
+    """
+    settings = get_settings_service().settings
+    if not settings.rate_limit_enabled:
+        return
+    limiter = http_request.app.state.limiter
+    limit_item = parse(f"{settings.public_flow_rate_limit_per_minute}/minute")
+    # hit() returns False once the window is exhausted for this (namespace, ip) key.
+    if not limiter._limiter.hit(limit_item, "public-workflow", limiter._key_func(http_request)):  # noqa: SLF001
+        raise HTTPException(
+            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+            detail={
+                "error": "Too many requests",
+                "code": "RATE_LIMITED",
+                "message": "Too many public workflow runs from this client. Please retry shortly.",
+            },
+        )
+
+
 @router.post(
     "",
     responses=WORKFLOW_EXECUTION_RESPONSES,
@@ -88,6 +115,10 @@ async def execute_public_workflow(
     # collected at import time.
     from langflow.api.v2.workflow import _stream_event_frames, _unknown_protocol_http_exception
 
+    # Throttle before any DB lookup or flow execution so an anonymous flood is
+    # rejected cheaply, not after spending the flow owner's resources.
+    _enforce_public_rate_limit(http_request)
+
     real_flow_id = UUID(request.flow_id)
 
     # Mirror v1 ``build_public_tmp`` error contract: blocked-component
diff --git a/src/backend/base/langflow/api/v2/workflow_reconstruction.py b/src/backend/base/langflow/api/v2/workflow_reconstruction.py
index b079de69b2..2d187ab070 100644
--- a/src/backend/base/langflow/api/v2/workflow_reconstruction.py
+++ b/src/backend/base/langflow/api/v2/workflow_reconstruction.py
@@ -10,9 +10,9 @@ from typing import TYPE_CHECKING
 
 from lfx.graph.graph.base import Graph
 from lfx.graph.schema import ResultData, RunOutputs
+from lfx.workflow.converters import run_response_to_workflow_response
 
 from langflow.api.v1.schemas import RunResponse
-from langflow.api.v2.converters import run_response_to_workflow_response
 from langflow.services.database.models.vertex_builds.crud import get_vertex_builds_by_job_id
 
 if TYPE_CHECKING:
@@ -20,6 +20,37 @@ if TYPE_CHECKING:
 
     from langflow.services.database.models.flow.model import FlowRead
 
+# Bound the structured search so a pathological/cyclic payload can't recurse
+# unboundedly; the session_id sits a few levels deep in the persisted Message.
+_SESSION_ID_SEARCH_MAX_DEPTH = 8
+
+
+def _recover_session_id(value: object, depth: int = 0) -> str | None:
+    """Find the session_id persisted in a terminal vertex_build's structured data.
+
+    The session is serialized inside the terminal Message (e.g.
+    ``results["message"]["data"]["session_id"]`` and the output rows) at a depth
+    that varies by component, so search the dict/list structure rather than
+    depending on one component's layout. Serialized text blobs are strings, not
+    walked, so table headers that merely contain the word "session_id" are ignored.
+    """
+    if depth > _SESSION_ID_SEARCH_MAX_DEPTH:
+        return None
+    if isinstance(value, dict):
+        candidate = value.get("session_id")
+        if isinstance(candidate, str) and candidate:
+            return candidate
+        for nested in value.values():
+            found = _recover_session_id(nested, depth + 1)
+            if found:
+                return found
+    elif isinstance(value, list):
+        for item in value:
+            found = _recover_session_id(item, depth + 1)
+            if found:
+                return found
+    return None
+
 
 async def reconstruct_workflow_response_from_job_id(
     session: AsyncSession,
@@ -66,8 +97,18 @@ async def reconstruct_workflow_response_from_job_id(
     # Convert vertex_build data to RunOutputs format
     run_outputs_list = [RunOutputs(inputs={}, outputs=[ResultData(**vb.data)]) for vb in terminal_vertex_builds]
 
+    # Recover the session_id the run executed under so GET status can continue the
+    # same chat/memory thread, instead of always reporting null. It is persisted
+    # inside the terminal Message (``results[...]["data"]["session_id"]``) at a
+    # depth that varies by component, so search the structured data. A data-only
+    # flow has no session to recover and correctly stays None.
+    session_id = next(
+        (sid for vb in terminal_vertex_builds if (sid := _recover_session_id(vb.data))),
+        None,
+    )
+
     # Create RunResponse and convert to WorkflowExecutionResponse
-    run_response = RunResponse(outputs=run_outputs_list, session_id=None)
+    run_response = RunResponse(outputs=run_outputs_list, session_id=session_id)
 
     return run_response_to_workflow_response(
         run_response=run_response,
diff --git a/src/backend/base/langflow/api/v2/workflow_validation.py b/src/backend/base/langflow/api/v2/workflow_validation.py
new file mode 100644
index 0000000000..31467193fb
--- /dev/null
+++ b/src/backend/base/langflow/api/v2/workflow_validation.py
@@ -0,0 +1,77 @@
+"""Request and permission guards for V2 workflow execution.
+
+These run before a flow executes: they reject unsupported field combinations,
+enforce owner-only overrides, apply the server-side component policy gate, and
+validate caller-supplied output selection. They raise ``HTTPException`` so the
+route handlers can surface a structured error without running the graph.
+"""
+
+from __future__ import annotations
+
+from fastapi import HTTPException, status
+from lfx.utils.flow_validation import CustomComponentValidationError, validate_flow_for_current_settings
+from lfx.workflow.converters import ParsedWorkflowRun
+
+from langflow.services.authorization.fetch import deny_to_404
+from langflow.services.database.models.flow.model import FlowRead
+from langflow.services.database.models.user.model import UserRead
+
+
+def _flow_not_found_privacy_exception(exc: HTTPException, flow_id: str) -> HTTPException:
+    return deny_to_404(exc, detail=f"Flow with id {flow_id} not found")
+
+
+def _reject_unsupported_sync_fields(parsed: ParsedWorkflowRun) -> None:
+    """Reject request fields the inline sync path does not execute."""
+    if parsed.mode != "sync":
+        return
+
+    unsupported_fields: list[str] = []
+    if parsed.data is not None:
+        unsupported_fields.append("data")
+    if parsed.files:
+        unsupported_fields.append("files")
+    if parsed.start_component_id is not None:
+        unsupported_fields.append("start_component_id")
+    if parsed.stop_component_id is not None:
+        unsupported_fields.append("stop_component_id")
+
+    if unsupported_fields:
+        fields = ", ".join(unsupported_fields)
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail={
+                "error": "Unsupported sync request fields",
+                "code": "SYNC_MODE_UNSUPPORTED_FIELDS",
+                "message": f"mode='sync' does not support request fields: {fields}. Use mode='stream' or "
+                "mode='background' for live-canvas overrides, files, or partial-run boundaries.",
+                "fields": unsupported_fields,
+            },
+        )
+
+
+def _enforce_flow_data_override_owner(parsed: ParsedWorkflowRun, flow: FlowRead, current_user: UserRead) -> None:
+    """Only the flow owner may execute caller-supplied graph data."""
+    if parsed.data is None or flow.user_id == current_user.id:
+        return
+
+    raise _flow_not_found_privacy_exception(
+        HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Only the flow owner can override flow data during execution",
+        ),
+        parsed.flow_id,
+    )
+
+
+def _validate_flow_data_for_execution(parsed: ParsedWorkflowRun, flow: FlowRead) -> None:
+    """Apply the same server-side component policy gate used by v1/public runs."""
+    try:
+        if parsed.data is not None:
+            validate_flow_for_current_settings(parsed.data)
+        elif flow.data:
+            validate_flow_for_current_settings(flow.data)
+    except CustomComponentValidationError as exc:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
+    except RuntimeError as exc:
+        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(exc)) from exc
diff --git a/src/backend/base/langflow/services/background_execution/runner.py b/src/backend/base/langflow/services/background_execution/runner.py
index 070d0645e9..be8fbd270b 100644
--- a/src/backend/base/langflow/services/background_execution/runner.py
+++ b/src/backend/base/langflow/services/background_execution/runner.py
@@ -31,7 +31,8 @@ from langflow.services.jobs.exceptions import HUMAN_INPUT_REQUIRED_EVENT, PauseR
 if TYPE_CHECKING:
     from uuid import UUID
 
-    from langflow.api.v2.adapters import StreamAdapter
+    from lfx.workflow.adapters import StreamAdapter
+
     from langflow.services.background_execution.live_bus import InMemoryLiveBus
     from langflow.services.jobs.service import JobService
 
diff --git a/src/backend/base/langflow/services/background_execution/service.py b/src/backend/base/langflow/services/background_execution/service.py
index 7ab87f605a..5c22189e1c 100644
--- a/src/backend/base/langflow/services/background_execution/service.py
+++ b/src/backend/base/langflow/services/background_execution/service.py
@@ -504,7 +504,7 @@ class BackgroundExecutionService(Service):
         return meta.get("stream_protocol") or "langflow"
 
     def _build_adapter(self, request: dict[str, Any], job_id: UUID, flow_id: UUID):
-        from langflow.api.v2.adapters import StreamAdapterContext, get_stream_adapter
+        from lfx.workflow.adapters import StreamAdapterContext, get_stream_adapter
 
         protocol = request.get("stream_protocol", "langflow")
         return get_stream_adapter(
diff --git a/src/backend/tests/unit/api/v2/test_output_event_parity.py b/src/backend/tests/unit/api/v2/test_output_event_parity.py
index e24eb6f69d..fdd50df35e 100644
--- a/src/backend/tests/unit/api/v2/test_output_event_parity.py
+++ b/src/backend/tests/unit/api/v2/test_output_event_parity.py
@@ -15,9 +15,9 @@ import json
 from unittest.mock import Mock
 
 from langflow.api.build import _output_meta_for_vertex
-from langflow.api.v2.adapters import StreamAdapterContext
-from langflow.api.v2.adapters.langflow import LangflowAdapter
-from langflow.api.v2.converters import build_component_output, resolve_output_type
+from lfx.workflow.adapters import StreamAdapterContext
+from lfx.workflow.adapters.langflow import LangflowAdapter
+from lfx.workflow.converters import build_component_output, resolve_output_type
 
 
 def _adapter() -> LangflowAdapter:
diff --git a/src/backend/tests/unit/api/v2/test_workflow_agui.py b/src/backend/tests/unit/api/v2/test_workflow_agui.py
index 22dcb5f0b5..7832b1cbac 100644
--- a/src/backend/tests/unit/api/v2/test_workflow_agui.py
+++ b/src/backend/tests/unit/api/v2/test_workflow_agui.py
@@ -12,12 +12,17 @@ Execution mode (``mode`` field):
 """
 
 import json
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
 from uuid import uuid4
 
 import pytest
+from fastapi import HTTPException
 from httpx import AsyncClient
 from langflow.services.database.models.flow.model import Flow
+from lfx.schema.workflow import WorkflowRunRequest
 from lfx.services.deps import session_scope
+from lfx.utils.flow_validation import CustomComponentValidationError
 
 
 def _agui_body(flow_id, *, message: str = "hello", mode: str = "sync", tweaks: dict | None = None) -> dict:
@@ -1388,3 +1393,137 @@ class TestStopWorkflowEndToEnd:
                 break
             await _asyncio.sleep(0.1)
         assert final in terminal, f"run did not terminalize after stop: got {final}"
+
+
+class TestV2WorkflowAdmission:
+    """Route-level admission guards: sync-field rejection, owner-override, RBAC masking, policy gate."""
+
+    async def test_sync_mode_rejects_live_canvas_only_fields(
+        self,
+        client: AsyncClient,
+        created_api_key,
+        empty_flow,
+    ):
+        """Sync runs must reject fields that only stream/background executes."""
+        body = _agui_body(empty_flow, mode="sync")
+        body.update(
+            {
+                "data": {"nodes": [], "edges": []},
+                "files": ["tmp/upload.txt"],
+                "start_component_id": "input-1",
+                "stop_component_id": "output-1",
+            }
+        )
+
+        response = await client.post(
+            "api/v2/workflows",
+            json=body,
+            headers={"x-api-key": created_api_key.api_key},
+        )
+
+        assert response.status_code == 422
+        detail = response.json()["detail"]
+        assert detail["code"] == "SYNC_MODE_UNSUPPORTED_FIELDS"
+        assert detail["fields"] == ["data", "files", "start_component_id", "stop_component_id"]
+
+    async def test_non_owner_data_override_is_hidden_as_404(self, monkeypatch: pytest.MonkeyPatch):
+        """Execute-only sharees must not inject alternate graph data into shared flows."""
+        from langflow.api.v2 import workflow as workflow_module
+
+        flow_id = uuid4()
+        owner_id = uuid4()
+        caller_id = uuid4()
+        flow = SimpleNamespace(
+            id=flow_id,
+            user_id=owner_id,
+            workspace_id=None,
+            folder_id=None,
+            data={"nodes": [], "edges": []},
+            name="shared",
+        )
+        current_user = SimpleNamespace(id=caller_id)
+
+        monkeypatch.setattr(workflow_module, "get_flow_by_id_or_endpoint_name", AsyncMock(return_value=flow))
+        monkeypatch.setattr(workflow_module, "ensure_flow_permission", AsyncMock(return_value=None))
+
+        with pytest.raises(HTTPException) as exc_info:
+            await workflow_module.execute_workflow(
+                WorkflowRunRequest(
+                    flow_id=str(flow_id),
+                    input_value="hi",
+                    mode="stream",
+                    data={"nodes": [], "edges": []},
+                ),
+                background_tasks=SimpleNamespace(),
+                http_request=SimpleNamespace(),
+                current_user=current_user,
+            )
+
+        assert exc_info.value.status_code == 404
+        assert exc_info.value.detail["code"] == "FLOW_NOT_FOUND"
+
+    async def test_execute_permission_denial_is_hidden_as_404(self, monkeypatch: pytest.MonkeyPatch):
+        """A denied share-aware fetch must not leak flow existence as a raw 403."""
+        from langflow.api.v2 import workflow as workflow_module
+
+        flow_id = uuid4()
+        caller_id = uuid4()
+        flow = SimpleNamespace(
+            id=flow_id,
+            user_id=uuid4(),
+            workspace_id=None,
+            folder_id=None,
+            data={"nodes": [], "edges": []},
+            name="private",
+        )
+
+        async def _deny(*_args, **_kwargs):
+            raise HTTPException(status_code=403, detail="denied")
+
+        monkeypatch.setattr(workflow_module, "get_flow_by_id_or_endpoint_name", AsyncMock(return_value=flow))
+        monkeypatch.setattr(workflow_module, "ensure_flow_permission", _deny)
+
+        with pytest.raises(HTTPException) as exc_info:
+            await workflow_module.execute_workflow(
+                WorkflowRunRequest(flow_id=str(flow_id), input_value="hi", mode="stream"),
+                background_tasks=SimpleNamespace(),
+                http_request=SimpleNamespace(),
+                current_user=SimpleNamespace(id=caller_id),
+            )
+
+        assert exc_info.value.status_code == 404
+        assert exc_info.value.detail["code"] == "FLOW_NOT_FOUND"
+
+    async def test_private_route_applies_component_policy_gate(self, monkeypatch: pytest.MonkeyPatch):
+        """The authenticated v2 route must run server-side component policy validation."""
+        from langflow.api.v2 import workflow as workflow_module
+        from langflow.api.v2 import workflow_validation as wf_val
+
+        flow_id = uuid4()
+        flow = SimpleNamespace(
+            id=flow_id,
+            user_id=uuid4(),
+            workspace_id=None,
+            folder_id=None,
+            data={"nodes": [], "edges": []},
+            name="private",
+        )
+
+        def _reject(_flow_data):
+            message = "custom components are disabled"
+            raise CustomComponentValidationError(message)
+
+        monkeypatch.setattr(workflow_module, "get_flow_by_id_or_endpoint_name", AsyncMock(return_value=flow))
+        monkeypatch.setattr(workflow_module, "ensure_flow_permission", AsyncMock(return_value=None))
+        monkeypatch.setattr(wf_val, "validate_flow_for_current_settings", _reject)
+
+        with pytest.raises(HTTPException) as exc_info:
+            await workflow_module.execute_workflow(
+                WorkflowRunRequest(flow_id=str(flow_id), input_value="hi", mode="stream"),
+                background_tasks=SimpleNamespace(),
+                http_request=SimpleNamespace(),
+                current_user=SimpleNamespace(id=flow.user_id),
+            )
+
+        assert exc_info.value.status_code == 400
+        assert exc_info.value.detail == "custom components are disabled"
diff --git a/src/backend/tests/unit/api/v2/test_workflow_public.py b/src/backend/tests/unit/api/v2/test_workflow_public.py
index 6d1e6653b7..3753317a4e 100644
--- a/src/backend/tests/unit/api/v2/test_workflow_public.py
+++ b/src/backend/tests/unit/api/v2/test_workflow_public.py
@@ -19,6 +19,7 @@ the mitigations the v2 public endpoint is supposed to inherit from v1:
 from __future__ import annotations
 
 from typing import TYPE_CHECKING, Any
+from uuid import uuid4
 
 import pytest
 from httpx import AsyncClient, codes
@@ -422,3 +423,30 @@ async def test_public_endpoint_surfaces_value_error_as_400(client: AsyncClient,
 
     assert response.status_code == codes.BAD_REQUEST
     assert response.json().get("detail") == "custom gate failure"
+
+
+def test_public_endpoint_throttles_per_ip(monkeypatch):
+    """The unauthenticated public endpoint throttles per client IP.
+
+    Each run executes as the flow owner (real cost), so an anonymous caller must
+    not be able to spin up unbounded runs. With the limit set to 2/min, the third
+    request from the same IP is rejected at the throttle (429) before any flow work.
+    """
+    from fastapi.testclient import TestClient
+    from langflow.main import create_app
+    from langflow.services.deps import get_settings_service
+
+    settings = get_settings_service().settings
+    monkeypatch.setattr(settings, "rate_limit_enabled", True)
+    monkeypatch.setattr(settings, "public_flow_rate_limit_per_minute", 2)
+
+    app = create_app()
+    sync_client = TestClient(app)
+    body = {"flow_id": str(uuid4()), "input_value": "hi"}
+
+    statuses = [sync_client.post("api/v2/workflows/public", json=body).status_code for _ in range(3)]
+
+    # First two pass the throttle (and fail downstream on the nonexistent flow);
+    # the third exhausts the 2/min window and is rejected at the throttle.
+    assert statuses[2] == codes.TOO_MANY_REQUESTS, statuses
+    assert codes.TOO_MANY_REQUESTS not in statuses[:2], statuses
diff --git a/src/backend/tests/unit/api/v2/test_workflow_reconstruction.py b/src/backend/tests/unit/api/v2/test_workflow_reconstruction.py
index f3b959ccb1..8dda2b93a1 100644
--- a/src/backend/tests/unit/api/v2/test_workflow_reconstruction.py
+++ b/src/backend/tests/unit/api/v2/test_workflow_reconstruction.py
@@ -12,12 +12,41 @@ from unittest.mock import MagicMock, patch
 from uuid import uuid4
 
 import pytest
-from langflow.api.v2.workflow_reconstruction import reconstruct_workflow_response_from_job_id
+from langflow.api.v2.workflow_reconstruction import _recover_session_id, reconstruct_workflow_response_from_job_id
 from langflow.services.database.models.vertex_builds.model import VertexBuildTable
 from lfx.interface.components import component_cache
 from lfx.utils.flow_validation import CustomComponentValidationError
 
 
+class TestRecoverSessionId:
+    """``_recover_session_id`` digs the persisted session_id out of terminal data.
+
+    The session_id sits inside the serialized terminal Message at a depth that
+    varies by component, so the search walks the dict/list structure and ignores
+    serialized text blobs that merely contain the word ``session_id``.
+    """
+
+    def test_recovers_session_id_nested_like_persisted_message(self):
+        # Mirrors the real shape: results["message"]["data"]["session_id"].
+        data = {"results": {"message": {"text_key": "text", "data": {"session_id": "thread-1", "text": "hi"}}}}
+        assert _recover_session_id(data) == "thread-1"
+
+    def test_recovers_session_id_from_list_of_rows(self):
+        data = {"outputs": {"dataframe": {"message": [{"text": "hi", "session_id": "thread-9"}]}}}
+        assert _recover_session_id(data) == "thread-9"
+
+    def test_ignores_session_id_appearing_only_inside_serialized_text(self):
+        # A rendered table header is a string, not a dict key, so it is not matched.
+        data = {"outputs": {"prompt": {"message": "| sender | session_id | text |"}}}
+        assert _recover_session_id(data) is None
+
+    def test_returns_none_for_data_only_flow(self):
+        assert _recover_session_id({"results": {}, "outputs": {"data": {"value": 1}}}) is None
+
+    def test_blank_session_id_is_not_recovered(self):
+        assert _recover_session_id({"data": {"session_id": ""}}) is None
+
+
 class TestWorkflowReconstruction:
     """Unit tests for workflow reconstruction logic."""
 
diff --git a/src/backend/tests/unit/background_execution/test_agui_replay_byte_identical.py b/src/backend/tests/unit/background_execution/test_agui_replay_byte_identical.py
index 8a1309d1aa..7037db3e38 100644
--- a/src/backend/tests/unit/background_execution/test_agui_replay_byte_identical.py
+++ b/src/backend/tests/unit/background_execution/test_agui_replay_byte_identical.py
@@ -16,8 +16,8 @@ import json
 
 import pytest
 from fastapi.sse import format_sse_event
-from langflow.api.v2.adapters import StreamAdapterContext, get_stream_adapter
 from langflow.services.background_execution.service import BackgroundExecutionService
+from lfx.workflow.adapters import StreamAdapterContext, get_stream_adapter
 
 pytestmark = pytest.mark.real_services
 
diff --git a/src/backend/tests/unit/background_execution/test_facade_real_services.py b/src/backend/tests/unit/background_execution/test_facade_real_services.py
index 89299370eb..a5879ce80c 100644
--- a/src/backend/tests/unit/background_execution/test_facade_real_services.py
+++ b/src/backend/tests/unit/background_execution/test_facade_real_services.py
@@ -15,10 +15,10 @@ from typing import TYPE_CHECKING
 from uuid import uuid4
 
 import pytest
-from langflow.api.v2.adapters import StreamAdapterContext, get_stream_adapter
 from langflow.services.background_execution.live_bus import InMemoryLiveBus, LiveFrame
 from langflow.services.background_execution.runner import JobRunner
 from langflow.services.database.models.jobs.model import JobStatus, SignalType
+from lfx.workflow.adapters import StreamAdapterContext, get_stream_adapter
 
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator
diff --git a/src/backend/tests/unit/background_execution/test_head_to_head_deltas.py b/src/backend/tests/unit/background_execution/test_head_to_head_deltas.py
index 35cf05ef52..bdc5d18fe5 100644
--- a/src/backend/tests/unit/background_execution/test_head_to_head_deltas.py
+++ b/src/backend/tests/unit/background_execution/test_head_to_head_deltas.py
@@ -27,11 +27,11 @@ import threading
 from uuid import uuid4
 
 import pytest
-from langflow.api.v2.adapters import StreamAdapterContext, get_stream_adapter
 from langflow.services.background_execution.executor import InProcessExecutor
 from langflow.services.background_execution.live_bus import InMemoryLiveBus
 from langflow.services.background_execution.runner import JobRunner
 from langflow.services.database.models.jobs.model import JobStatus
+from lfx.workflow.adapters import StreamAdapterContext, get_stream_adapter
 
 pytestmark = pytest.mark.real_services
 
diff --git a/src/backend/tests/unit/background_execution/test_last_event_id_contract.py b/src/backend/tests/unit/background_execution/test_last_event_id_contract.py
index b44aefa665..0ee556a137 100644
--- a/src/backend/tests/unit/background_execution/test_last_event_id_contract.py
+++ b/src/backend/tests/unit/background_execution/test_last_event_id_contract.py
@@ -23,9 +23,9 @@ from uuid import uuid4
 
 import pytest
 from fastapi.sse import format_sse_event
-from langflow.api.v2.adapters import StreamAdapterContext, get_stream_adapter
 from langflow.services.background_execution.live_bus import InMemoryLiveBus
 from langflow.services.background_execution.runner import JobRunner
+from lfx.workflow.adapters import StreamAdapterContext, get_stream_adapter
 
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator
diff --git a/src/backend/tests/unit/background_execution/test_multi_worker_boot_real_services.py b/src/backend/tests/unit/background_execution/test_multi_worker_boot_real_services.py
index 35fb3af161..46d48e4dfb 100644
--- a/src/backend/tests/unit/background_execution/test_multi_worker_boot_real_services.py
+++ b/src/backend/tests/unit/background_execution/test_multi_worker_boot_real_services.py
@@ -18,10 +18,10 @@ from typing import TYPE_CHECKING
 from uuid import uuid4
 
 import pytest
-from langflow.api.v2.adapters import StreamAdapterContext, get_stream_adapter
 from langflow.services.background_execution.live_bus import InMemoryLiveBus
 from langflow.services.background_execution.runner import JobRunner
 from langflow.services.database.models.jobs.model import JobStatus
+from lfx.workflow.adapters import StreamAdapterContext, get_stream_adapter
 
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator
diff --git a/src/backend/tests/unit/services/background_execution/test_runner.py b/src/backend/tests/unit/services/background_execution/test_runner.py
index ab15be1809..b1003247e4 100644
--- a/src/backend/tests/unit/services/background_execution/test_runner.py
+++ b/src/backend/tests/unit/services/background_execution/test_runner.py
@@ -14,11 +14,11 @@ from typing import TYPE_CHECKING
 from uuid import uuid4
 
 import pytest
-from langflow.api.v2.adapters import StreamAdapterContext, get_stream_adapter
 from langflow.services.background_execution.live_bus import InMemoryLiveBus
 from langflow.services.background_execution.runner import JobRunner
 from langflow.services.database.models.jobs.model import JobStatus
 from langflow.services.deps import get_job_service
+from lfx.workflow.adapters import StreamAdapterContext, get_stream_adapter
 
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator
diff --git a/src/lfx/src/lfx/schema/workflow.py b/src/lfx/src/lfx/schema/workflow.py
index 22f95e8384..17caf2b43a 100644
--- a/src/lfx/src/lfx/schema/workflow.py
+++ b/src/lfx/src/lfx/schema/workflow.py
@@ -18,6 +18,14 @@ GLOBAL_VALUE_MAX_LEN = 64 * 1024
 GlobalVarKey = Annotated[str, StringConstraints(min_length=1, max_length=GLOBAL_KEY_MAX_LEN)]
 GlobalVarValue = Annotated[str, StringConstraints(max_length=GLOBAL_VALUE_MAX_LEN)]
 
+# Bounds on the unauthenticated public-flow request, so an anonymous caller
+# can't post arbitrarily large strings that are held in memory and persisted to
+# MessageTable. input_value is a chat message (same 64 KB ceiling as a global
+# variable value); session_id is namespaced under the visitor's virtual flow id,
+# so a key-sized bound is ample.
+PublicInputValue = Annotated[str, StringConstraints(max_length=GLOBAL_VALUE_MAX_LEN)]
+PublicSessionId = Annotated[str, StringConstraints(max_length=GLOBAL_KEY_MAX_LEN)]
+
 
 class JobStatus(str, Enum):
     """Job execution status."""
@@ -123,7 +131,8 @@ class WorkflowExecutionRequest(BaseModel):
             "Keys may use any printable string up to "
             f"{GLOBAL_KEY_MAX_LEN} chars; values are capped at "
             f"{GLOBAL_VALUE_MAX_LEN} chars. Body globals always win over the "
-            "legacy ``X-LANGFLOW-GLOBAL-VAR-*`` headers."
+            "legacy ``X-LANGFLOW-GLOBAL-VAR-*`` headers. Honored in sync mode; "
+            "ignored for stream/background modes."
         ),
     )
 
@@ -246,7 +255,8 @@ class WorkflowRunRequest(BaseModel):
             "Keys may use any printable string up to "
             f"{GLOBAL_KEY_MAX_LEN} chars; values are capped at "
             f"{GLOBAL_VALUE_MAX_LEN} chars. Body globals always win over the "
-            "legacy ``X-LANGFLOW-GLOBAL-VAR-*`` headers."
+            "legacy ``X-LANGFLOW-GLOBAL-VAR-*`` headers. Honored in sync mode; "
+            "ignored for stream/background modes."
         ),
     )
     idempotency_key: str | None = Field(
@@ -318,8 +328,8 @@ class PublicWorkflowRunRequest(BaseModel):
     """
 
     flow_id: str = Field(..., description="UUID of the public flow to run.")
-    input_value: str = Field("", description="Chat-style input value.")
-    session_id: str | None = Field(
+    input_value: PublicInputValue = Field("", description="Chat-style input value.")
+    session_id: PublicSessionId | None = Field(
         None,
         description=("Optional caller session. Always namespaced under the visitor's virtual flow id by the endpoint."),
     )
diff --git a/src/lfx/src/lfx/services/settings/groups/runtime.py b/src/lfx/src/lfx/services/settings/groups/runtime.py
index d2ec3680da..fb22607c9a 100644
--- a/src/lfx/src/lfx/services/settings/groups/runtime.py
+++ b/src/lfx/src/lfx/services/settings/groups/runtime.py
@@ -102,6 +102,11 @@ class RuntimeSettings(BaseModel):
     worker_timeout: int = 300
     """Timeout for the API calls in seconds."""
 
+    workflow_execution_timeout: int = 300
+    """Wall-clock ceiling in seconds for a single v2 workflow run, applied to every
+    execution mode. Sync runs raise a 408; stream, background, and public runs emit
+    the protocol's terminal-error event and (for background) mark the job failed."""
+
     public_flow_cleanup_interval: int = Field(default=3600, gt=600)
     """The interval in seconds at which public temporary flows will be cleaned up.
     Default is 1 hour (3600 seconds). Minimum is 600 seconds (10 minutes)."""
diff --git a/src/lfx/src/lfx/services/settings/groups/security.py b/src/lfx/src/lfx/services/settings/groups/security.py
index fecd02b8c0..e55cef584a 100644
--- a/src/lfx/src/lfx/services/settings/groups/security.py
+++ b/src/lfx/src/lfx/services/settings/groups/security.py
@@ -78,6 +78,10 @@ class SecuritySettings(BaseModel):
     """Storage backend for rate limiting. Use 'memory://' for single-server or 'redis://host:port' for multi-server."""
     rate_limit_trust_proxy: bool = False
     """Trust X-Forwarded-For header when behind a reverse proxy. Only enable when behind a trusted proxy."""
+    public_flow_rate_limit_per_minute: int = 20
+    """Public-flow runs allowed per minute per IP on the unauthenticated /api/v2/workflows/public endpoint.
+    Each run executes as the flow owner (real CPU/DB/LLM-credit cost), so anonymous callers are throttled
+    separately from (and more generously than) the login limit. Gated by rate_limit_enabled."""
 
     @field_validator("cors_origins", mode="before")
     @classmethod
diff --git a/src/lfx/src/lfx/workflow/__init__.py b/src/lfx/src/lfx/workflow/__init__.py
new file mode 100644
index 0000000000..5f37f03283
--- /dev/null
+++ b/src/lfx/src/lfx/workflow/__init__.py
@@ -0,0 +1,14 @@
+"""V2 workflow contract layer shared by the langflow backend and ``lfx serve``.
+
+This package holds the protocol-agnostic pieces of the V2 workflow API:
+
+    - ``adapters``: the ``StreamAdapter`` protocol, the ``langflow``/``agui`` SSE
+      adapters, and the registry (``get_stream_adapter`` / ``available_protocols``).
+    - ``agui_translator``: translates EventManager events into AG-UI events.
+    - ``converters``: parses ``WorkflowRunRequest`` and builds the structured
+      ``WorkflowExecutionResponse``.
+
+It depends only on ``lfx.schema.workflow`` and ``ag_ui`` (no langflow imports),
+so both runtimes consume one contract. The langflow backend layers its stateful
+"vehicle" (database, job queue, auth) on top.
+"""
diff --git a/src/backend/base/langflow/api/v2/adapters/__init__.py b/src/lfx/src/lfx/workflow/adapters/__init__.py
similarity index 93%
rename from src/backend/base/langflow/api/v2/adapters/__init__.py
rename to src/lfx/src/lfx/workflow/adapters/__init__.py
index 9a670666bc..de5e94a17e 100644
--- a/src/backend/base/langflow/api/v2/adapters/__init__.py
+++ b/src/lfx/src/lfx/workflow/adapters/__init__.py
@@ -11,12 +11,16 @@ Adding a new protocol is one new module under ``adapters/`` plus one
 
 from __future__ import annotations
 
-from collections.abc import Callable, Iterable
+from collections.abc import Callable
 from dataclasses import dataclass
-from typing import Any, ClassVar, Protocol, runtime_checkable
+from typing import TYPE_CHECKING, Protocol, runtime_checkable
 
 from pydantic import BaseModel
 
+if TYPE_CHECKING:
+    from collections.abc import Iterable
+    from typing import Any, ClassVar
+
 
 @dataclass(frozen=True)
 class StreamEvent:
@@ -127,5 +131,5 @@ def available_protocols() -> list[str]:
 
 # Built-in adapter registrations happen here, after the registry is defined.
 # Import for side-effect: each module calls ``register_stream_adapter``.
-from langflow.api.v2.adapters import agui as _agui  # noqa: E402, F401
-from langflow.api.v2.adapters import langflow as _langflow  # noqa: E402, F401
+from lfx.workflow.adapters import agui as _agui  # noqa: E402, F401
+from lfx.workflow.adapters import langflow as _langflow  # noqa: E402, F401
diff --git a/src/backend/base/langflow/api/v2/adapters/agui.py b/src/lfx/src/lfx/workflow/adapters/agui.py
similarity index 84%
rename from src/backend/base/langflow/api/v2/adapters/agui.py
rename to src/lfx/src/lfx/workflow/adapters/agui.py
index 742faee3d2..9febad33f0 100644
--- a/src/backend/base/langflow/api/v2/adapters/agui.py
+++ b/src/lfx/src/lfx/workflow/adapters/agui.py
@@ -7,17 +7,19 @@ Per-run state lives on the wrapped translator instance.
 
 from __future__ import annotations
 
-from collections.abc import Iterable
-from typing import Any, ClassVar
+from typing import TYPE_CHECKING, Any, ClassVar
 
-from ag_ui.core import BaseEvent
-
-from langflow.api.v2.adapters import (
+from lfx.workflow.adapters import (
     StreamAdapterContext,
     StreamEvent,
     register_stream_adapter,
 )
-from langflow.api.v2.agui_translator import AGUITranslator
+from lfx.workflow.agui_translator import AGUITranslator
+
+if TYPE_CHECKING:
+    from collections.abc import Iterable
+
+    from ag_ui.core import BaseEvent
 
 # Durable AG-UI milestones. ``TEXT_MESSAGE_CONTENT`` is the per-token delta and
 # is ephemeral; the START/END lifecycle frames around it are durable so a
@@ -80,9 +82,10 @@ class AGUIAdapter:
         return [_to_stream_event(e) for e in self._translator.translate("error", {"error": str(error)})]
 
     def cancel_events(self, reason: str) -> Iterable[StreamEvent]:
-        # AG-UI has no cancellation primitive in the local event model; route
-        # through the translator so any open text lifecycle is closed first.
-        return [_to_stream_event(e) for e in self._translator.translate("error", {"error": reason})]
+        # AG-UI has no cancel primitive: close any open text lifecycle, emit a
+        # CUSTOM cancel marker, then RUN_FINISHED. A user-stop must not replay as
+        # RUN_ERROR, which clients read as a genuine failure.
+        return [_to_stream_event(e) for e in self._translator.cancel(reason)]
 
     @property
     def terminal_error_type(self) -> str | None:
diff --git a/src/backend/base/langflow/api/v2/adapters/langflow.py b/src/lfx/src/lfx/workflow/adapters/langflow.py
similarity index 87%
rename from src/backend/base/langflow/api/v2/adapters/langflow.py
rename to src/lfx/src/lfx/workflow/adapters/langflow.py
index 69004e0d2b..b00efa4467 100644
--- a/src/backend/base/langflow/api/v2/adapters/langflow.py
+++ b/src/lfx/src/lfx/workflow/adapters/langflow.py
@@ -8,17 +8,18 @@ clients (curl users, the v1 frontend) can read it without learning anything new.
 from __future__ import annotations
 
 import json
-from collections.abc import Iterable
-from typing import Any, ClassVar
+from typing import TYPE_CHECKING, Any, ClassVar
 
 from lfx.schema.workflow import OutputEvent
-
-from langflow.api.v2.adapters import (
+from lfx.workflow.adapters import (
     StreamAdapterContext,
     StreamEvent,
     register_stream_adapter,
 )
-from langflow.api.v2.converters import build_component_output, resolve_output_type
+from lfx.workflow.converters import build_component_output, resolve_output_type
+
+if TYPE_CHECKING:
+    from collections.abc import Iterable
 
 # Durable milestones for the langflow wire protocol. ``token`` is the only
 # high-volume ephemeral type; everything else the build loop emits is a
@@ -93,6 +94,7 @@ class LangflowAdapter:
             output_type=resolve_output_type(output_meta.get("output_types"), output_meta.get("vertex_type")),
             display_name=output_meta.get("display_name"),
             result_data=build_data.get("data"),
+            valid=bool(build_data.get("valid", True)),
         )
         output = OutputEvent(component_id=component_id, **component_output.model_dump())
         payload = {"event": "output", "data": output.model_dump(mode="json")}
@@ -106,8 +108,11 @@ class LangflowAdapter:
         return (StreamEvent(type="error", data_json=json.dumps(payload)),)
 
     def cancel_events(self, reason: str) -> Iterable[StreamEvent]:
-        payload = {"event": "error", "data": {"error": reason}}
-        return (StreamEvent(type="error", data_json=json.dumps(payload)),)
+        # A deliberate user stop is its own terminal, not an ``error``: a
+        # re-attaching client must be able to tell a cancel apart from a genuine
+        # failure instead of seeing the stream end in ``error``.
+        payload = {"event": "cancelled", "data": {"reason": reason}}
+        return (StreamEvent(type="cancelled", data_json=json.dumps(payload)),)
 
     @property
     def terminal_error_type(self) -> str | None:
diff --git a/src/backend/base/langflow/api/v2/agui_translator.py b/src/lfx/src/lfx/workflow/agui_translator.py
similarity index 97%
rename from src/backend/base/langflow/api/v2/agui_translator.py
rename to src/lfx/src/lfx/workflow/agui_translator.py
index 60d46f14b3..434796623c 100644
--- a/src/backend/base/langflow/api/v2/agui_translator.py
+++ b/src/lfx/src/lfx/workflow/agui_translator.py
@@ -156,6 +156,19 @@ class AGUITranslator:
             return events
         return []
 
+    def cancel(self, reason: str) -> list[BaseEvent]:
+        """Close any open text lifecycle, mark the run cancelled, then finish.
+
+        AG-UI has no cancel primitive, so a deliberate stop is surfaced as a
+        namespaced CUSTOM marker (generic clients ignore it) followed by a clean
+        ``RUN_FINISHED`` rather than ``RUN_ERROR``, which clients read as a
+        genuine failure.
+        """
+        events = self._drain_messages()
+        events.append(CustomEvent(name="langflow.run.cancelled", value={"reason": reason}))
+        events.append(RunFinishedEvent(run_id=self.run_id, thread_id=self.thread_id))
+        return events
+
     def _translate_token(self, data: dict) -> list[BaseEvent]:
         """Map a ``token`` event to text-message events.
 
diff --git a/src/backend/base/langflow/api/v2/converters.py b/src/lfx/src/lfx/workflow/converters.py
similarity index 96%
rename from src/backend/base/langflow/api/v2/converters.py
rename to src/lfx/src/lfx/workflow/converters.py
index d51a86a525..a69dd6b1f3 100644
--- a/src/backend/base/langflow/api/v2/converters.py
+++ b/src/lfx/src/lfx/workflow/converters.py
@@ -24,7 +24,7 @@ from __future__ import annotations
 
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Protocol
 
 from lfx.schema.workflow import (
     ComponentOutput,
@@ -41,7 +41,18 @@ from lfx.schema.workflow import (
 if TYPE_CHECKING:
     from lfx.graph.graph.base import Graph
 
-    from langflow.api.v1.schemas import RunResponse
+
+class RunResponseLike(Protocol):
+    """Structural type for a v1-style run response.
+
+    Defined here so this module never imports from langflow. The backend's
+    ``RunResponse`` (a list of ``outputs`` plus a ``session_id``) satisfies it
+    structurally, and lfx's own run path can supply any object with the same
+    shape.
+    """
+
+    outputs: list[Any] | None
+    session_id: str | None
 
 
 @dataclass(frozen=True)
@@ -259,8 +270,6 @@ def _simplify_output_content(content: Any, output_type: str) -> Any:
             result_data = _extract_nested_value(content, *path)
             if result_data is not None:
                 return result_data
-    # TODO: Future scope - Add dataframe-specific extraction logic
-    # The following code is commented out pending further requirements analysis:
     if output_type == "dataframe":
         # For dataframe types, try multiple path combinations in order
         dataframe_paths = [
@@ -337,6 +346,7 @@ def build_component_output(
     output_type: str,
     display_name: str | None,
     result_data: Any,
+    valid: bool = True,
 ) -> ComponentOutput:
     """Build a ``ComponentOutput`` from one component's result data.
 
@@ -363,7 +373,6 @@ def build_component_output(
             # Non-output nodes:
             # - For data types: extract and show content
             # - For message types: extract metadata only (source, file_path)
-            # TODO: Future scope - Add support for "dataframe" output type
             if output_type in ["data", "dataframe"]:
                 content = _simplify_output_content(raw_content, output_type)
             else:
@@ -381,9 +390,15 @@ def build_component_output(
             if isinstance(result_metadata, dict):
                 metadata.update(result_metadata)
 
+    # Derive per-component status instead of hardcoding COMPLETED so a client
+    # trusting it is not given a false positive: a build that produced an error
+    # artifact (``output_type == "error"``) or that the graph marked invalid is
+    # FAILED. This mirrors the agui translator, which already keys node status on
+    # the same ``valid`` flag, so sync and both stream protocols agree.
+    status = JobStatus.FAILED if (output_type == "error" or not valid) else JobStatus.COMPLETED
     return ComponentOutput(
         type=output_type,
-        status=JobStatus.COMPLETED,
+        status=status,
         display_name=resolved_display_name,
         content=content,
         metadata=metadata,
@@ -504,7 +519,7 @@ def workflow_response_from_output_events(
 
 
 def run_response_to_workflow_response(
-    run_response: RunResponse,
+    run_response: RunResponseLike,
     flow_id: str,
     job_id: str,
     inputs: dict[str, Any],
diff --git a/src/lfx/tests/unit/services/settings/test_settings_composition.py b/src/lfx/tests/unit/services/settings/test_settings_composition.py
index af48b814bc..f8f035b8ee 100644
--- a/src/lfx/tests/unit/services/settings/test_settings_composition.py
+++ b/src/lfx/tests/unit/services/settings/test_settings_composition.py
@@ -133,6 +133,7 @@ EXPECTED_FIELDS = {
     "dev",
     "event_delivery",
     "worker_timeout",
+    "workflow_execution_timeout",
     "public_flow_cleanup_interval",
     "public_flow_expiration",
     "webhook_polling_interval",
@@ -161,6 +162,7 @@ EXPECTED_FIELDS = {
     "rate_limit_per_minute",
     "rate_limit_storage_uri",
     "rate_limit_trust_proxy",
+    "public_flow_rate_limit_per_minute",
     "custom_component_admin_only",
     "allow_components_paths_override",
     # RuntimeSettings
diff --git a/src/backend/tests/unit/api/v2/adapters/__init__.py b/src/lfx/tests/unit/workflow/__init__.py
similarity index 100%
rename from src/backend/tests/unit/api/v2/adapters/__init__.py
rename to src/lfx/tests/unit/workflow/__init__.py
diff --git a/src/lfx/tests/unit/workflow/adapters/__init__.py b/src/lfx/tests/unit/workflow/adapters/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/src/backend/tests/unit/api/v2/adapters/test_agui_adapter.py b/src/lfx/tests/unit/workflow/adapters/test_agui_adapter.py
similarity index 94%
rename from src/backend/tests/unit/api/v2/adapters/test_agui_adapter.py
rename to src/lfx/tests/unit/workflow/adapters/test_agui_adapter.py
index 83ce814199..90a8356f55 100644
--- a/src/backend/tests/unit/api/v2/adapters/test_agui_adapter.py
+++ b/src/lfx/tests/unit/workflow/adapters/test_agui_adapter.py
@@ -10,7 +10,7 @@ from __future__ import annotations
 import json
 
 import pytest
-from langflow.api.v2.adapters import (
+from lfx.workflow.adapters import (
     StreamAdapterContext,
     get_stream_adapter,
 )
@@ -131,17 +131,20 @@ class TestErrorEventsFallback:
         assert [event.type for event in events] == ["TEXT_MESSAGE_END", "RUN_ERROR"]
         assert json.loads(events[0].data_json)["messageId"] == "m1"
 
-    def test_cancel_events_close_open_text_message_before_run_error(self):
-        """Cancellation uses adapter-owned framing and preserves AG-UI lifecycle."""
+    def test_cancel_events_close_open_text_then_custom_marker_and_run_finished(self):
+        """A user-stop emits a CUSTOM cancel marker + RUN_FINISHED (never RUN_ERROR), after closing open text."""
         adapter = get_stream_adapter("agui", _ctx())
         list(adapter.initial_events())
         list(adapter.translate("token", {"id": "m1", "chunk": "partial"}))
 
         events = list(adapter.cancel_events("cancelled"))
 
-        assert [event.type for event in events] == ["TEXT_MESSAGE_END", "RUN_ERROR"]
+        assert [event.type for event in events] == ["TEXT_MESSAGE_END", "CUSTOM", "RUN_FINISHED"]
+        assert all(event.type != "RUN_ERROR" for event in events)
         assert json.loads(events[0].data_json)["messageId"] == "m1"
-        assert json.loads(events[1].data_json)["message"] == "cancelled"
+        marker = json.loads(events[1].data_json)
+        assert marker["name"] == "langflow.run.cancelled"
+        assert marker["value"]["reason"] == "cancelled"
 
 
 class TestTerminalErrorType:
diff --git a/src/backend/tests/unit/api/v2/adapters/test_event_durability.py b/src/lfx/tests/unit/workflow/adapters/test_event_durability.py
similarity index 95%
rename from src/backend/tests/unit/api/v2/adapters/test_event_durability.py
rename to src/lfx/tests/unit/workflow/adapters/test_event_durability.py
index 39f82cbfa1..f5eafbcad6 100644
--- a/src/backend/tests/unit/api/v2/adapters/test_event_durability.py
+++ b/src/lfx/tests/unit/workflow/adapters/test_event_durability.py
@@ -8,7 +8,7 @@ volume, only useful live.
 
 from __future__ import annotations
 
-from langflow.api.v2.adapters import StreamAdapterContext, get_stream_adapter
+from lfx.workflow.adapters import StreamAdapterContext, get_stream_adapter
 
 
 def _ctx() -> StreamAdapterContext:
diff --git a/src/backend/tests/unit/api/v2/adapters/test_human_input_event.py b/src/lfx/tests/unit/workflow/adapters/test_human_input_event.py
similarity index 97%
rename from src/backend/tests/unit/api/v2/adapters/test_human_input_event.py
rename to src/lfx/tests/unit/workflow/adapters/test_human_input_event.py
index f0a06f29e6..012ee132e5 100644
--- a/src/backend/tests/unit/api/v2/adapters/test_human_input_event.py
+++ b/src/lfx/tests/unit/workflow/adapters/test_human_input_event.py
@@ -9,7 +9,7 @@ from __future__ import annotations
 
 import json
 
-from langflow.api.v2.adapters import StreamAdapterContext, get_stream_adapter
+from lfx.workflow.adapters import StreamAdapterContext, get_stream_adapter
 
 _PAYLOAD = {
     "request_id": "node:job-1",
diff --git a/src/backend/tests/unit/api/v2/adapters/test_langflow_adapter.py b/src/lfx/tests/unit/workflow/adapters/test_langflow_adapter.py
similarity index 92%
rename from src/backend/tests/unit/api/v2/adapters/test_langflow_adapter.py
rename to src/lfx/tests/unit/workflow/adapters/test_langflow_adapter.py
index 3b32d1812c..ab28288cb3 100644
--- a/src/backend/tests/unit/api/v2/adapters/test_langflow_adapter.py
+++ b/src/lfx/tests/unit/workflow/adapters/test_langflow_adapter.py
@@ -10,7 +10,7 @@ from __future__ import annotations
 import json
 
 import pytest
-from langflow.api.v2.adapters import (
+from lfx.workflow.adapters import (
     StreamAdapterContext,
     get_stream_adapter,
 )
@@ -62,12 +62,13 @@ class TestErrorHandling:
         assert payload == {"event": "error", "data": {"error": "boom"}}
         assert evt.type == "error"
 
-    def test_cancel_events_emit_protocol_error_payload(self):
+    def test_cancel_events_emit_cancelled_payload_not_error(self):
+        """A user-stop is its own ``cancelled`` terminal, not ``error``, so a client can tell it from a failure."""
         adapter = get_stream_adapter("langflow", _ctx())
         [evt] = list(adapter.cancel_events("cancelled"))
         payload = json.loads(evt.data_json)
-        assert payload == {"event": "error", "data": {"error": "cancelled"}}
-        assert evt.type == "error"
+        assert payload == {"event": "cancelled", "data": {"reason": "cancelled"}}
+        assert evt.type == "cancelled"
 
     def test_terminal_error_type_is_error(self):
         """Used by the buffer task to decide JobStatus.FAILED."""
diff --git a/src/backend/tests/unit/api/v2/adapters/test_registry.py b/src/lfx/tests/unit/workflow/adapters/test_registry.py
similarity index 99%
rename from src/backend/tests/unit/api/v2/adapters/test_registry.py
rename to src/lfx/tests/unit/workflow/adapters/test_registry.py
index 04b2c22b3e..cd0f72a5f7 100644
--- a/src/backend/tests/unit/api/v2/adapters/test_registry.py
+++ b/src/lfx/tests/unit/workflow/adapters/test_registry.py
@@ -9,7 +9,7 @@ registers under a string name; the endpoint dispatches by name and returns
 from __future__ import annotations
 
 import pytest
-from langflow.api.v2.adapters import (
+from lfx.workflow.adapters import (
     STREAM_ADAPTERS,
     StreamAdapterContext,
     StreamEvent,
diff --git a/src/backend/tests/unit/api/v2/test_agui_translator.py b/src/lfx/tests/unit/workflow/test_agui_translator.py
similarity index 99%
rename from src/backend/tests/unit/api/v2/test_agui_translator.py
rename to src/lfx/tests/unit/workflow/test_agui_translator.py
index cba1efe63a..9286ac3466 100644
--- a/src/backend/tests/unit/api/v2/test_agui_translator.py
+++ b/src/lfx/tests/unit/workflow/test_agui_translator.py
@@ -25,7 +25,7 @@ from ag_ui.core import (
     ToolCallResultEvent,
     ToolCallStartEvent,
 )
-from langflow.api.v2.agui_translator import AGUITranslator
+from lfx.workflow.agui_translator import AGUITranslator
 
 
 def test_run_lifecycle_emits_started_and_finished():
diff --git a/src/backend/tests/unit/api/v2/test_converters.py b/src/lfx/tests/unit/workflow/test_converters.py
similarity index 96%
rename from src/backend/tests/unit/api/v2/test_converters.py
rename to src/lfx/tests/unit/workflow/test_converters.py
index eaa684016b..830fc2ca7d 100644
--- a/src/backend/tests/unit/api/v2/test_converters.py
+++ b/src/lfx/tests/unit/workflow/test_converters.py
@@ -27,7 +27,15 @@ from unittest.mock import Mock
 from uuid import uuid4
 
 import pytest
-from langflow.api.v2.converters import (
+from lfx.schema.workflow import (
+    ComponentOutput,
+    ErrorDetail,
+    JobStatus,
+    OutputReason,
+    WorkflowExecutionResponse,
+    WorkflowJobResponse,
+)
+from lfx.workflow.converters import (
     _build_metadata_for_non_output,
     _extract_file_path,
     _extract_model_source,
@@ -42,14 +50,6 @@ from langflow.api.v2.converters import (
     create_job_response,
     run_response_to_workflow_response,
 )
-from lfx.schema.workflow import (
-    ComponentOutput,
-    ErrorDetail,
-    JobStatus,
-    OutputReason,
-    WorkflowExecutionResponse,
-    WorkflowJobResponse,
-)
 
 
 def _setup_graph_get_vertex(graph: Mock, vertices: list[Mock]) -> None:
@@ -1127,8 +1127,8 @@ class TestParseWorkflowRunRequest:
     """``parse_workflow_run_request`` projects ``WorkflowRunRequest`` onto ``ParsedWorkflowRun``."""
 
     def test_minimal_body_round_trips_with_defaults(self):
-        from langflow.api.v2.converters import parse_workflow_run_request
         from lfx.schema.workflow import WorkflowRunRequest
+        from lfx.workflow.converters import parse_workflow_run_request
 
         parsed = parse_workflow_run_request(WorkflowRunRequest(flow_id=_VALID_UUID))
 
@@ -1144,8 +1144,8 @@ class TestParseWorkflowRunRequest:
         assert parsed.files is None
 
     def test_full_body_round_trip(self):
-        from langflow.api.v2.converters import parse_workflow_run_request
         from lfx.schema.workflow import WorkflowMode, WorkflowRunRequest
+        from lfx.workflow.converters import parse_workflow_run_request
 
         request = WorkflowRunRequest(
             flow_id=_VALID_UUID,
@@ -1174,8 +1174,8 @@ class TestParseWorkflowRunRequest:
 
     def test_run_id_is_always_none_on_the_parsed_record(self):
         """The endpoint generates run_id; callers cannot supply it via the body."""
-        from langflow.api.v2.converters import parse_workflow_run_request
         from lfx.schema.workflow import WorkflowRunRequest
+        from lfx.workflow.converters import parse_workflow_run_request
 
         parsed = parse_workflow_run_request(WorkflowRunRequest(flow_id=_VALID_UUID))
         assert parsed.run_id is None
@@ -1630,6 +1630,54 @@ class TestBuildComponentOutput:
         assert output.content is None
         assert output.metadata == {"component_type": "ChatOutput"}
 
+    def test_error_output_type_yields_failed_status(self):
+        # An error artifact carries type "error"; per-component status must reflect
+        # the failure instead of the hardcoded COMPLETED a client would trust.
+        output = build_component_output(
+            component_id="ChatOutput-abc",
+            is_output=True,
+            vertex_type="ChatOutput",
+            output_type="error",
+            display_name="Chat Output",
+            result_data=None,
+        )
+        assert output.status == JobStatus.FAILED
+
+    def test_invalid_build_yields_failed_status(self):
+        # The stream path passes the vertex ``valid`` flag (the agui translator keys
+        # node status on the same signal); an invalid build is FAILED, not COMPLETED.
+        output = build_component_output(
+            component_id="ChatOutput-abc",
+            is_output=True,
+            vertex_type="ChatOutput",
+            output_type="message",
+            display_name="Chat Output",
+            result_data=None,
+            valid=False,
+        )
+        assert output.status == JobStatus.FAILED
+
+
+class TestGlobalsSyncOnlyDocumented:
+    """The ``globals`` field is honored on sync only; the schema must say so.
+
+    Stream/background converge on ``generate_flow_events`` which never receives
+    request globals, so the field description documents the sync-only limitation
+    the same way ``output_ids`` already documents "Ignored for stream/background".
+    """
+
+    def test_workflow_run_request_globals_documents_sync_only(self):
+        from lfx.schema.workflow import WorkflowRunRequest
+
+        description = WorkflowRunRequest.model_fields["globals"].description
+        assert "ignored for stream/background" in description.lower()
+
+    def test_workflow_execution_request_globals_documents_sync_only(self):
+        from lfx.schema.workflow import WorkflowExecutionRequest
+
+        description = WorkflowExecutionRequest.model_fields["globals"].description
+        assert "ignored for stream/background" in description.lower()
+
 
 if __name__ == "__main__":
     pytest.main([__file__, "-v"])