chat.py

from __future__ import annotations

import asyncio
import re
import time
import traceback
import uuid
from typing import TYPE_CHECKING, Annotated

from fastapi import APIRouter, BackgroundTasks, Body, Depends, HTTPException, Request, status
from fastapi.responses import StreamingResponse
from lfx.graph.graph.base import Graph
from lfx.graph.utils import log_vertex_build
from lfx.log.logger import logger
from lfx.schema.schema import InputValueRequest, OutputValue
from lfx.services.cache.utils import CacheMiss
from lfx.utils.flow_validation import (
    CustomComponentValidationError,
    validate_flow_for_current_settings,
    validate_public_flow_no_code_execution,
)
from sqlmodel import select

from langflow.api.build import cancel_flow_build, get_flow_events_response, start_flow_build
from langflow.api.limited_background_tasks import LimitVertexBuildBackgroundTasks
from langflow.api.utils import (
    CurrentActiveUser,
    DbSession,
    EventDeliveryType,
    build_and_cache_graph_from_data,
    build_graph_from_db,
    format_elapsed_time,
    format_exception_message,
    get_top_level_vertices,
    parse_exception,
    scope_session_to_namespace,
    verify_public_flow_and_get_user,
)
from langflow.api.v1.schemas import (
    CancelFlowResponse,
    FlowDataRequest,
    ResultDataResponse,
    StreamData,
    VertexBuildResponse,
    VerticesOrderResponse,
)
from langflow.exceptions.component import ComponentBuildError
from langflow.services.auth.utils import get_current_active_user, get_current_user_optional
from langflow.services.authorization import FlowAction, ensure_flow_permission
from langflow.services.authorization.fetch import deny_to_404
from langflow.services.chat.service import ChatService
from langflow.services.database.models.flow.model import AccessTypeEnum, Flow
from langflow.services.database.models.user.model import User
from langflow.services.deps import (
    get_chat_service,
    get_queue_service,
    get_settings_service,
    get_telemetry_service,
    session_scope,
)
from langflow.services.job_queue.service import JobQueueNotFoundError, JobQueueService
from langflow.services.telemetry.schema import ComponentPayload, PlaygroundPayload

if TYPE_CHECKING:
    from lfx.graph.vertex.vertex_types import InterfaceVertex

router = APIRouter(tags=["Chat"])


async def _verify_job_ownership(job_id: str, current_user: CurrentActiveUser, queue_service: JobQueueService) -> None:
    """Raise HTTP 404 if the requesting user does not own the job.

    Jobs with no registered owner (build_public_tmp) are accessible to any authenticated user.
    """
    job_owner = await queue_service.get_job_owner(job_id)
    if job_owner is not None and job_owner != current_user.id:
        await logger.awarning(
            "Ownership check failed: user %s tried to access job %s owned by %s",
            current_user.id,
            job_id,
            job_owner,
        )
        raise HTTPException(status_code=404, detail=f"Job not found: {job_id}")


@router.post(
    "/build/{flow_id}/vertices",
    deprecated=True,
    include_in_schema=False,
)
async def retrieve_vertices_order(
    *,
    flow_id: uuid.UUID,
    background_tasks: BackgroundTasks,
    data: Annotated[FlowDataRequest | None, Body(embed=True)] | None = None,
    stop_component_id: str | None = None,
    start_component_id: str | None = None,
    session: DbSession,
    current_user: CurrentActiveUser,
) -> VerticesOrderResponse:
    """Retrieve the vertices order for a given flow.

    Args:
        flow_id (str): The ID of the flow.
        background_tasks (BackgroundTasks): The background tasks.
        data (Optional[FlowDataRequest], optional): The flow data. Defaults to None.
        stop_component_id (str, optional): The ID of the stop component. Defaults to None.
        start_component_id (str, optional): The ID of the start component. Defaults to None.
        session (AsyncSession, optional): The session dependency.
        current_user: The authenticated user (required so the handler can
            run the same authorization guard the supported /build/{flow_id}/flow
            route uses).

    Returns:
        VerticesOrderResponse: The response containing the ordered vertex IDs and the run ID.

    Raises:
        HTTPException: If there is an error checking the build status.
    """
    # Owner-or-public ownership check + ensure_flow_permission(EXECUTE) — same
    # pattern as build_flow below. ``build_graph_from_db`` reaches into the DB
    # with a bare ``session.get(Flow, flow_id)`` that has no owner filter, so
    # without this gate any authenticated user could build any other user's
    # flow by guessing a UUID. Even though the route is deprecated and hidden
    # from the schema, it remains routed and reachable.
    stmt = (
        select(Flow)
        .where(Flow.id == flow_id)
        .where((Flow.user_id == current_user.id) | (Flow.access_type == AccessTypeEnum.PUBLIC))
    )
    flow = (await session.exec(stmt)).first()
    if not flow:
        raise HTTPException(status_code=404, detail=f"Flow with id {flow_id} not found")
    await ensure_flow_permission(
        current_user,
        FlowAction.EXECUTE,
        flow_id=flow_id,
        flow_user_id=flow.user_id,
        workspace_id=flow.workspace_id,
        folder_id=flow.folder_id,
    )

    chat_service = get_chat_service()
    telemetry_service = get_telemetry_service()
    start_time = time.perf_counter()
    components_count = None
    run_id = str(uuid.uuid4())
    try:
        if not data:
            graph = await build_graph_from_db(flow_id=flow_id, session=session, chat_service=chat_service)
        else:
            graph = await build_and_cache_graph_from_data(
                flow_id=flow_id, graph_data=data.model_dump(), chat_service=chat_service
            )
        graph = graph.prepare(stop_component_id, start_component_id)
        graph.set_run_id(run_id)

        # Now vertices is a list of lists
        # We need to get the id of each vertex
        # and return the same structure but only with the ids
        components_count = len(graph.vertices)
        vertices_to_run = list(graph.vertices_to_run.union(get_top_level_vertices(graph, graph.vertices_to_run)))
        await chat_service.set_cache(str(flow_id), graph)
        background_tasks.add_task(
            telemetry_service.log_package_playground,
            PlaygroundPayload(
                playground_seconds=int(time.perf_counter() - start_time),
                playground_component_count=components_count,
                playground_success=True,
                playground_run_id=run_id,
            ),
        )
        return VerticesOrderResponse(ids=graph.first_layer, run_id=graph.run_id, vertices_to_run=vertices_to_run)
    except Exception as exc:
        background_tasks.add_task(
            telemetry_service.log_package_playground,
            PlaygroundPayload(
                playground_seconds=int(time.perf_counter() - start_time),
                playground_component_count=components_count,
                playground_success=False,
                playground_error_message=str(exc),
                playground_run_id=run_id,
            ),
        )
        if "stream or streaming set to True" in str(exc):
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        if isinstance(exc, CustomComponentValidationError):
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        await logger.aexception("Error checking build status")
        raise HTTPException(status_code=500, detail=str(exc)) from exc


@router.post("/build/{flow_id}/flow")
async def build_flow(
    *,
    flow_id: uuid.UUID,
    background_tasks: LimitVertexBuildBackgroundTasks,
    inputs: Annotated[InputValueRequest | None, Body(embed=True)] = None,
    data: Annotated[FlowDataRequest | None, Body(embed=True)] = None,
    files: list[str] | None = None,
    stop_component_id: str | None = None,
    start_component_id: str | None = None,
    log_builds: bool = True,
    current_user: CurrentActiveUser,
    queue_service: Annotated[JobQueueService, Depends(get_queue_service)],
    flow_name: str | None = None,
    event_delivery: EventDeliveryType = EventDeliveryType.POLLING,
):
    """Build and process a flow, returning a job ID for event polling.

    This endpoint requires authentication through the CurrentActiveUser dependency.
    For public flows that don't require authentication, use the /build_public_tmp/flow_id/flow endpoint.

    Args:
        flow_id: UUID of the flow to build
        background_tasks: Background tasks manager
        inputs: Optional input values for the flow
        data: Optional flow data
        files: Optional files to include
        stop_component_id: Optional ID of component to stop at
        start_component_id: Optional ID of component to start from
        log_builds: Whether to log the build process
        current_user: The authenticated user
        queue_service: Queue service for job management
        flow_name: Optional name for the flow
        event_delivery: Optional event delivery type - default is streaming

    Returns:
        Dict with job_id that can be used to poll for build status
    """
    # Share-aware load: when the authorization plugin signals cross-user fetch
    # support, the row loads by id alone and the plugin decides. Otherwise we
    # keep the historical owner-or-PUBLIC scoping so the OSS pass-through
    # default cannot widen visibility. PUBLIC flows stay buildable by any
    # authenticated user in both modes.
    from langflow.api.v1.flows_helpers import _read_flow

    async with session_scope() as session:
        flow = await _read_flow(session, flow_id, current_user.id)
        if flow is None:
            public_stmt = select(Flow).where(
                Flow.id == flow_id,
                Flow.access_type == AccessTypeEnum.PUBLIC,
            )
            flow = (await session.exec(public_stmt)).first()
        if not flow:
            await logger.awarning(
                "Flow access denied for user %s: flow %s not found or not owned",
                current_user.id,
                flow_id,
            )
            raise HTTPException(status_code=404, detail=f"Flow with id {flow_id} not found")

    # Authorize the execute action — runs the authorization plugin if registered,
    # no-op in OSS pass-through. Audited regardless. A plugin deny becomes 404
    # so the response is identical to "flow does not exist" and the caller
    # cannot enumerate UUIDs by probing for 403 vs 404.
    try:
        await ensure_flow_permission(
            current_user,
            FlowAction.EXECUTE,
            flow_id=flow_id,
            flow_user_id=flow.user_id,
            workspace_id=flow.workspace_id,
            folder_id=flow.folder_id,
        )
    except HTTPException as exc:
        raise deny_to_404(exc, detail=f"Flow with id {flow_id} not found") from exc

    # Execute-only shares must run the stored graph — non-owners cannot inject
    # alternate flow data via the request body (would bypass the owner's definition).
    if data is not None and flow.user_id != current_user.id:
        raise deny_to_404(
            HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Only the flow owner can override flow data during build",
            ),
            detail=f"Flow with id {flow_id} not found",
        )

    try:
        if data:
            validate_flow_for_current_settings(data.model_dump())
        elif flow and flow.data:
            validate_flow_for_current_settings(flow.data)
    except CustomComponentValidationError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    except RuntimeError as exc:
        raise HTTPException(status_code=503, detail=str(exc)) from exc

    job_id = await start_flow_build(
        flow_id=flow_id,
        background_tasks=background_tasks,
        inputs=inputs,
        data=data,
        files=files,
        stop_component_id=stop_component_id,
        start_component_id=start_component_id,
        log_builds=log_builds,
        current_user=current_user,
        queue_service=queue_service,
        flow_name=flow_name,
    )
    await queue_service.register_job_owner(job_id, current_user.id)

    # This is required to support FE tests - we need to be able to set the event delivery to direct
    if event_delivery != EventDeliveryType.DIRECT:
        return {"job_id": job_id}
    return await get_flow_events_response(
        job_id=job_id,
        queue_service=queue_service,
        event_delivery=event_delivery,
    )


@router.get("/build/{job_id}/events")
async def get_build_events(
    job_id: str,
    queue_service: Annotated[JobQueueService, Depends(get_queue_service)],
    current_user: CurrentActiveUser,
    *,
    event_delivery: EventDeliveryType = EventDeliveryType.STREAMING,
):
    """Get events for a specific build job.

    Requires authentication and ownership verification. A job owner is registered
    when build_flow is called; if a registered owner does not match the requesting
    user the endpoint returns 404 to avoid leaking job existence.
    Jobs started via build_public_tmp have no registered owner and remain accessible
    to any authenticated user.
    """
    await _verify_job_ownership(job_id, current_user, queue_service)
    return await get_flow_events_response(
        job_id=job_id,
        queue_service=queue_service,
        event_delivery=event_delivery,
    )


@router.post(
    "/build/{job_id}/cancel",
    response_model=CancelFlowResponse,
)
async def cancel_build(
    job_id: str,
    queue_service: Annotated[JobQueueService, Depends(get_queue_service)],
    current_user: CurrentActiveUser,
):
    """Cancel a specific build job.

    Requires authentication and ownership verification to prevent a user from
    aborting another user's running build (DoS via job cancellation).
    Jobs with no registered owner (build_public_tmp) are accessible to any
    authenticated user, consistent with get_build_events.
    """
    await _verify_job_ownership(job_id, current_user, queue_service)
    try:
        # Cancel the flow build and check if it was successful
        cancellation_success = await cancel_flow_build(job_id=job_id, queue_service=queue_service)

        if cancellation_success:
            # Cancellation succeeded or wasn't needed
            return CancelFlowResponse(success=True, message="Flow build cancelled successfully")
        # Cancellation was attempted but failed
        return CancelFlowResponse(success=False, message="Failed to cancel flow build")
    except asyncio.CancelledError:
        # If CancelledError reaches here, it means the task was not successfully cancelled
        await logger.aerror(f"Failed to cancel flow build for job_id {job_id} (CancelledError caught)")
        return CancelFlowResponse(success=False, message="Failed to cancel flow build")
    except ValueError as exc:
        # Job not found
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
    except JobQueueNotFoundError as exc:
        await logger.aerror(f"Job not found: {job_id}. Error: {exc!s}")
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Job not found: {exc!s}") from exc
    except Exception as exc:
        # Any other unexpected error
        await logger.aexception(f"Error cancelling flow build for job_id {job_id}: {exc}")
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc)) from exc


@router.post("/build/{flow_id}/vertices/{vertex_id}", deprecated=True, include_in_schema=False)
async def build_vertex(
    *,
    flow_id: uuid.UUID,
    vertex_id: str,
    background_tasks: BackgroundTasks,
    inputs: Annotated[InputValueRequest | None, Body(embed=True)] = None,
    files: list[str] | None = None,
    current_user: CurrentActiveUser,
) -> VertexBuildResponse:
    """Build a vertex instead of the entire graph.

    Args:
        flow_id (str): The ID of the flow.
        vertex_id (str): The ID of the vertex to build.
        background_tasks (BackgroundTasks): The background tasks dependency.
        inputs (Optional[InputValueRequest], optional): The input values for the vertex. Defaults to None.
        files (List[str], optional): The files to use. Defaults to None.
        current_user (Any, optional): The current user dependency. Defaults to Depends(get_current_active_user).

    Returns:
        VertexBuildResponse: The response containing the built vertex information.

    Raises:
        HTTPException: If there is an error building the vertex.

    """
    # Owner-or-public ownership check + ensure_flow_permission(EXECUTE) — same
    # pattern as retrieve_vertices_order above. The route is deprecated and
    # hidden from the schema but still routed, and ``build_graph_from_db``
    # loads the flow with no owner filter, so without this gate any
    # authenticated user could build a vertex on someone else's flow.
    async with session_scope() as authz_session:
        stmt = (
            select(Flow)
            .where(Flow.id == flow_id)
            .where((Flow.user_id == current_user.id) | (Flow.access_type == AccessTypeEnum.PUBLIC))
        )
        flow = (await authz_session.exec(stmt)).first()
    if not flow:
        raise HTTPException(status_code=404, detail=f"Flow with id {flow_id} not found")
    await ensure_flow_permission(
        current_user,
        FlowAction.EXECUTE,
        flow_id=flow_id,
        flow_user_id=flow.user_id,
        workspace_id=flow.workspace_id,
        folder_id=flow.folder_id,
    )

    chat_service = get_chat_service()
    telemetry_service = get_telemetry_service()
    flow_id_str = str(flow_id)

    next_runnable_vertices = []
    top_level_vertices = []
    start_time = time.perf_counter()
    error_message = None
    run_id = None
    try:
        graph: Graph = await chat_service.get_cache(flow_id_str)
    except KeyError as exc:
        raise HTTPException(status_code=404, detail="Graph not found") from exc

    try:
        cache = await chat_service.get_cache(flow_id_str)
        if isinstance(cache, CacheMiss):
            # If there's no cache
            await logger.awarning(f"No cache found for {flow_id_str}. Building graph starting at {vertex_id}")
            async with session_scope() as session:
                graph = await build_graph_from_db(
                    flow_id=flow_id,
                    session=session,
                    chat_service=chat_service,
                )
            run_id = str(uuid.uuid4())
            graph.set_run_id(run_id)
        else:
            graph = cache.get("result")
            await graph.initialize_run()
            run_id = graph.run_id
        vertex = graph.get_vertex(vertex_id)

        try:
            lock = chat_service.async_cache_locks[flow_id_str]
            vertex_build_result = await graph.build_vertex(
                vertex_id=vertex_id,
                user_id=str(current_user.id),
                inputs_dict=inputs.model_dump() if inputs else {},
                files=files,
                get_cache=chat_service.get_cache,
                set_cache=chat_service.set_cache,
            )
            result_dict = vertex_build_result.result_dict
            params = vertex_build_result.params
            valid = vertex_build_result.valid
            artifacts = vertex_build_result.artifacts
            next_runnable_vertices = await graph.get_next_runnable_vertices(lock, vertex=vertex, cache=False)
            top_level_vertices = graph.get_top_level_vertices(next_runnable_vertices)
            result_data_response = ResultDataResponse.model_validate(result_dict, from_attributes=True)
        except Exception as exc:  # noqa: BLE001
            if isinstance(exc, ComponentBuildError):
                params = exc.message
                tb = exc.formatted_traceback
            else:
                tb = traceback.format_exc()
                await logger.aexception("Error building Component")
                params = format_exception_message(exc)
            message = {"errorMessage": params, "stackTrace": tb}
            valid = False
            error_message = params
            output_label = vertex.outputs[0]["name"] if vertex.outputs else "output"
            outputs = {output_label: OutputValue(message=message, type="error")}
            result_data_response = ResultDataResponse(results={}, outputs=outputs)
            artifacts = {}
            background_tasks.add_task(graph.end_all_traces_in_context(error=exc))
            # If there's an error building the vertex
            # we need to clear the cache
            await chat_service.clear_cache(flow_id_str)

        result_data_response.message = artifacts

        # Log the vertex build
        if not vertex.will_stream:
            background_tasks.add_task(
                log_vertex_build,
                flow_id=flow_id_str,
                vertex_id=vertex_id,
                valid=valid,
                params=params,
                data=result_data_response,
                artifacts=artifacts,
            )

        timedelta = time.perf_counter() - start_time

        duration = format_elapsed_time(timedelta)
        result_data_response.duration = duration
        result_data_response.timedelta = timedelta
        vertex.add_build_time(timedelta)
        inactivated_vertices = list(graph.inactivated_vertices)
        graph.reset_inactivated_vertices()
        graph.reset_activated_vertices()

        await chat_service.set_cache(flow_id_str, graph)

        # graph.stop_vertex tells us if the user asked
        # to stop the build of the graph at a certain vertex
        # if it is in next_vertices_ids, we need to remove other
        # vertices from next_vertices_ids
        if graph.stop_vertex and graph.stop_vertex in next_runnable_vertices:
            next_runnable_vertices = [graph.stop_vertex]

        if not graph.run_manager.vertices_being_run and not next_runnable_vertices:
            background_tasks.add_task(graph.end_all_traces_in_context())

        build_response = VertexBuildResponse(
            inactivated_vertices=list(set(inactivated_vertices)),
            next_vertices_ids=list(set(next_runnable_vertices)),
            top_level_vertices=list(set(top_level_vertices)),
            valid=valid,
            params=params,
            id=vertex.id,
            data=result_data_response,
        )
        background_tasks.add_task(
            telemetry_service.log_package_component,
            ComponentPayload(
                component_name=vertex_id.split("-")[0],
                component_id=vertex_id,
                component_seconds=int(time.perf_counter() - start_time),
                component_success=valid,
                component_error_message=error_message,
                component_run_id=run_id,
            ),
        )
    except Exception as exc:
        background_tasks.add_task(
            telemetry_service.log_package_component,
            ComponentPayload(
                component_name=vertex_id.split("-")[0],
                component_id=vertex_id,
                component_seconds=int(time.perf_counter() - start_time),
                component_success=False,
                component_error_message=str(exc),
                component_run_id=run_id if "run_id" in locals() else None,
            ),
        )
        if isinstance(exc, CustomComponentValidationError):
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        await logger.aexception("Error building Component")
        message = parse_exception(exc)
        raise HTTPException(status_code=500, detail=message) from exc

    return build_response


async def _stream_vertex(flow_id: str, vertex_id: str, chat_service: ChatService):
    graph = None
    try:
        try:
            cache = await chat_service.get_cache(flow_id)
        except Exception as exc:  # noqa: BLE001
            await logger.aexception("Error building Component")
            yield str(StreamData(event="error", data={"error": str(exc)}))
            return

        if isinstance(cache, CacheMiss):
            # If there's no cache
            msg = f"No cache found for {flow_id}."
            await logger.aerror(msg)
            yield str(StreamData(event="error", data={"error": msg}))
            return
        else:
            graph = cache.get("result")

        try:
            vertex: InterfaceVertex = graph.get_vertex(vertex_id)
        except Exception as exc:  # noqa: BLE001
            await logger.aexception("Error building Component")
            yield str(StreamData(event="error", data={"error": str(exc)}))
            return

        if not hasattr(vertex, "stream"):
            msg = f"Vertex {vertex_id} does not support streaming"
            await logger.aerror(msg)
            yield str(StreamData(event="error", data={"error": msg}))
            return

        if isinstance(vertex.built_result, str) and vertex.built_result:
            stream_data = StreamData(
                event="message",
                data={"message": f"Streaming vertex {vertex_id}"},
            )
            yield str(stream_data)
            stream_data = StreamData(
                event="message",
                data={"chunk": vertex.built_result},
            )
            yield str(stream_data)

        elif not vertex.frozen or not vertex.built:
            await logger.adebug(f"Streaming vertex {vertex_id}")
            stream_data = StreamData(
                event="message",
                data={"message": f"Streaming vertex {vertex_id}"},
            )
            yield str(stream_data)
            try:
                async for chunk in vertex.stream():
                    stream_data = StreamData(
                        event="message",
                        data={"chunk": chunk},
                    )
                    yield str(stream_data)
            except Exception as exc:  # noqa: BLE001
                await logger.aexception("Error building Component")
                exc_message = parse_exception(exc)
                if exc_message == "The message must be an iterator or an async iterator.":
                    exc_message = "This stream has already been closed."
                yield str(StreamData(event="error", data={"error": exc_message}))
        elif vertex.result is not None:
            stream_data = StreamData(
                event="message",
                data={"chunk": vertex.built_result},
            )
            yield str(stream_data)
        else:
            msg = f"No result found for vertex {vertex_id}"
            await logger.aerror(msg)
            yield str(StreamData(event="error", data={"error": msg}))
            return
    finally:
        await logger.adebug("Closing stream")
        if graph:
            await chat_service.set_cache(flow_id, graph)
        yield str(StreamData(event="close", data={"message": "Stream closed"}))


@router.get(
    "/build/{flow_id}/{vertex_id}/stream",
    response_class=StreamingResponse,
    deprecated=True,
    dependencies=[Depends(get_current_active_user)],
    include_in_schema=False,
)
async def build_vertex_stream(
    flow_id: uuid.UUID,
    vertex_id: str,
):
    """Build a vertex instead of the entire graph.

    This function is responsible for building a single vertex instead of the entire graph.
    It takes the `flow_id` and `vertex_id` as required parameters, and an optional `session_id`.
    It also depends on the `ChatService` and `SessionService` services.

    If `session_id` is not provided, it retrieves the graph from the cache using the `chat_service`.
    If `session_id` is provided, it loads the session data using the `session_service`.

    Once the graph is obtained, it retrieves the specified vertex using the `vertex_id`.
    If the vertex does not support streaming, an error is raised.
    If the vertex has a built result, it sends the result as a chunk.
    If the vertex is not frozen or not built, it streams the vertex data.
    If the vertex has a result, it sends the result as a chunk.
    If none of the above conditions are met, an error is raised.

    If any exception occurs during the process, an error message is sent.
    Finally, the stream is closed.

    Returns:
        A `StreamingResponse` object with the streamed vertex data in text/event-stream format.

    Raises:
        HTTPException: If an error occurs while building the vertex.
    """
    try:
        return StreamingResponse(
            _stream_vertex(str(flow_id), vertex_id, get_chat_service()),
            media_type="text/event-stream",
        )
    except Exception as exc:
        raise HTTPException(status_code=500, detail="Error building Component") from exc


async def build_flow_and_stream(flow_id, inputs, background_tasks, current_user):
    queue_service = get_queue_service()
    build_response = await build_flow(
        flow_id=flow_id,
        inputs=inputs,
        background_tasks=background_tasks,
        current_user=current_user,
        queue_service=queue_service,
        event_delivery=EventDeliveryType.STREAMING,
    )
    job_id = build_response["job_id"]
    return await get_flow_events_response(
        job_id=job_id,
        queue_service=queue_service,
        event_delivery=EventDeliveryType.STREAMING,
    )


# Public flow file paths must be `{source_flow_id}/{safe_basename}` — uploads
# under that namespace are the only legitimate inputs for an unauthenticated
# build. Anything else (absolute paths, traversal, foreign flow_ids) is a
# probe at the arbitrary-file-read class of bug.
_PUBLIC_FILE_PATH_RE = re.compile(
    r"^([0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12})/([^/\\]+)$"
)
_PUBLIC_FILE_REJECTED_SUBSTRINGS = ("\x00", "..", "\\")


def _validate_public_files(files: list[str] | None, source_flow_id: uuid.UUID) -> None:
    """Reject file references that aren't `{source_flow_id}/{basename}`."""
    if not files:
        return
    expected_flow_id = str(source_flow_id).lower()
    for entry in files:
        if not isinstance(entry, str) or not entry:
            raise HTTPException(status_code=400, detail="Invalid file entry")
        if any(token in entry for token in _PUBLIC_FILE_REJECTED_SUBSTRINGS):
            raise HTTPException(status_code=400, detail="Invalid file path")
        match = _PUBLIC_FILE_PATH_RE.match(entry)
        if not match:
            raise HTTPException(status_code=400, detail="Invalid file path format")
        flow_id_segment, basename = match.group(1), match.group(2)
        if flow_id_segment.lower() != expected_flow_id:
            raise HTTPException(status_code=400, detail="File not in this flow's namespace")
        if basename in (".", ".."):
            raise HTTPException(status_code=400, detail="Invalid filename")


@router.post("/build_public_tmp/{flow_id}/flow")
async def build_public_tmp(
    *,
    background_tasks: LimitVertexBuildBackgroundTasks,
    flow_id: uuid.UUID,
    inputs: Annotated[InputValueRequest | None, Body(embed=True)] = None,
    files: list[str] | None = None,
    stop_component_id: str | None = None,
    start_component_id: str | None = None,
    log_builds: bool | None = True,
    flow_name: str | None = None,
    request: Request,
    queue_service: Annotated[JobQueueService, Depends(get_queue_service)],
    authenticated_user: Annotated[User | None, Depends(get_current_user_optional)] = None,
    event_delivery: EventDeliveryType = EventDeliveryType.POLLING,
):
    """Build a public flow without requiring authentication.

    This endpoint is specifically for public flows that don't require authentication.
    It uses a client_id cookie to create a deterministic flow ID for tracking purposes.

    Security Note:
    - The 'data' parameter is NOT accepted to prevent flow definition tampering
    - Public flows must execute the stored flow definition only
    - The flow definition is always loaded from the database
    - Caller-supplied 'inputs.session' is namespaced under the (client_id,
      flow_id) virtual flow ID so an unauthenticated caller cannot address a
      session that lives outside its own namespace (CVE-2026-33017)

    The endpoint:
    1. Verifies the requested flow is marked as public in the database
    2. Creates a deterministic UUID based on client_id and flow_id
    3. Uses the flow owner's permissions to build the flow
    4. Always loads the flow definition from the database

    Requirements:
    - The flow must be marked as PUBLIC in the database
    - The request must include a client_id cookie

    Args:
        flow_id: UUID of the public flow to build
        background_tasks: Background tasks manager
        inputs: Optional input values for the flow
        files: Optional files to include
        stop_component_id: Optional ID of component to stop at
        start_component_id: Optional ID of component to start from
        log_builds: Whether to log the build process
        flow_name: Optional name for the flow
        request: FastAPI request object (needed for cookie access)
        queue_service: Queue service for job management
        authenticated_user: Optional authenticated user (resolved from cookie/token if present)
        event_delivery: Optional event delivery type - default is streaming

    Returns:
        Dict with job_id that can be used to poll for build status
    """
    try:
        # Reject caller-supplied file references that aren't scoped to this
        # public flow's own storage namespace. Done before any flow lookup so
        # malformed requests fail fast and don't touch the DB.
        _validate_public_files(files, flow_id)

        # Verify this is a public flow and get the associated user
        client_id = request.cookies.get("client_id")
        # Only use authenticated user_id when auto-login is disabled.
        # When AUTO_LOGIN=TRUE, the frontend uses client_id for UUID v5,
        # so the backend must match to avoid flow_id mismatch.
        auth_settings = get_settings_service().auth_settings
        authenticated_user_id = authenticated_user.id if authenticated_user and not auth_settings.AUTO_LOGIN else None
        owner_user, new_flow_id = await verify_public_flow_and_get_user(
            flow_id=flow_id,
            client_id=client_id,
            authenticated_user_id=authenticated_user_id,
        )

        # Defends CVE-2026-33017: scope caller session into the (client_id, flow_id) namespace.
        if inputs is not None and inputs.session is not None:
            scoped_session = scope_session_to_namespace(inputs.session, str(new_flow_id))
            if scoped_session != inputs.session:
                inputs = inputs.model_copy(update={"session": scoped_session})

        # Validate the stored flow data after the public-access boundary.
        # Public flows never accept client-supplied data.
        async with session_scope() as session:
            flow = await session.get(Flow, flow_id)
            if flow and flow.data:
                validate_flow_for_current_settings(flow.data)
                # Block unauthenticated builds of flows that run arbitrary code
                # (Python interpreter/REPL, legacy Python Code Structured tool,
                # Smart Transform lambda). Without this, any public flow
                # containing such a component is an unauthenticated server-side
                # code-execution primitive (report H1-3754930).
                validate_public_flow_no_code_execution(flow.data)

        # flow_id=new_flow_id for tracking/sessions/messages (virtual, per-user isolation).
        # source_flow_id=flow_id to load the actual flow data from the database.
        job_id = await start_flow_build(
            flow_id=new_flow_id,
            source_flow_id=flow_id,
            background_tasks=background_tasks,
            inputs=inputs,
            data=None,  # Always None - public flows load from database only
            files=files,
            stop_component_id=stop_component_id,
            start_component_id=start_component_id,
            log_builds=log_builds or False,
            current_user=owner_user,
            queue_service=queue_service,
            flow_name=flow_name or f"{authenticated_user_id or client_id}_{flow_id}",
        )
        # Gate the public events/cancel endpoints to jobs that were actually
        # started through this public build path, preventing unauthenticated
        # callers from reading or cancelling private-flow builds by job_id.
        await queue_service.register_public_job(job_id)
    except CustomComponentValidationError as exc:
        await logger.awarning(f"Public flow validation failed: {exc}")
        raise HTTPException(status_code=400, detail="This flow cannot be executed.") from exc
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    except Exception as exc:
        await logger.aexception("Error building public flow")
        if isinstance(exc, HTTPException):
            raise
        raise HTTPException(status_code=500, detail=str(exc)) from exc
    if event_delivery != EventDeliveryType.DIRECT:
        return {"job_id": job_id}
    return await get_flow_events_response(
        job_id=job_id,
        queue_service=queue_service,
        event_delivery=event_delivery,
    )


async def _assert_public_job(job_id: str, queue_service: JobQueueService) -> None:
    """Raise HTTP 404 if job_id was not registered through the public build endpoint.

    Prevents unauthenticated callers from reading or cancelling private-flow
    builds by guessing or leaking a job_id.

    Why 404 not 403: returning 403 would confirm the job exists under a different
    access tier, leaking information about private builds. 404 is neutral.
    """
    if not await queue_service.is_public_job_async(job_id):
        # Static detail — do not reflect job_id back; avoid confirming which IDs exist.
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job not found")


@router.get("/build_public_tmp/{job_id}/events")
async def get_build_events_public(
    job_id: str,
    queue_service: Annotated[JobQueueService, Depends(get_queue_service)],
    *,
    event_delivery: EventDeliveryType = EventDeliveryType.STREAMING,
):
    """Get events for a public flow build job.

    This endpoint does not require authentication, matching the public build endpoint.
    It is used by the shareable playground to consume build events.
    """
    await _assert_public_job(job_id, queue_service)
    return await get_flow_events_response(
        job_id=job_id,
        queue_service=queue_service,
        event_delivery=event_delivery,
    )


@router.post(
    "/build_public_tmp/{job_id}/cancel",
    response_model=CancelFlowResponse,
)
async def cancel_build_public(
    job_id: str,
    queue_service: Annotated[JobQueueService, Depends(get_queue_service)],
):
    """Cancel a public flow build job.

    This endpoint does not require authentication, matching the public build endpoint.
    It is used by the shareable playground to cancel builds.
    """
    await _assert_public_job(job_id, queue_service)
    try:
        cancellation_success = await cancel_flow_build(job_id=job_id, queue_service=queue_service)

        if cancellation_success:
            return CancelFlowResponse(success=True, message="Flow build cancelled successfully")
        return CancelFlowResponse(success=False, message="Failed to cancel flow build")
    except asyncio.CancelledError:
        await logger.aerror(f"Failed to cancel public flow build for job_id {job_id} (CancelledError caught)")
        return CancelFlowResponse(success=False, message="Failed to cancel flow build")
    except ValueError as exc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
    except JobQueueNotFoundError as exc:
        await logger.aerror(f"Public job not found: {job_id}. Error: {exc!s}")
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Job not found: {exc!s}") from exc
    except Exception as exc:
        await logger.aexception(f"Error cancelling public flow build for job_id {job_id}: {exc}")
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc)) from exc