From fa59cc50e46822b3bc71dfc66aa89fe665014f2e Mon Sep 17 00:00:00 2001 From: Janardan Singh Kavia Date: Wed, 21 Jan 2026 16:34:10 -0500 Subject: [PATCH] feat: Implement synchronous workflow execution API (#11255) * feat: Create controller shell and schema model for new workflow API - Add workflow API endpoints (POST /workflow, GET /workflow, POST /workflow/stop) - Implement developer API protection with settings check - Add comprehensive workflow schema models with proper validation - Create extensive unit test suite covering all scenarios - Apply Ruff linting standards and fix all code quality issues - Support API key authentication for all workflow endpoints * fix: Move developer API check to router-level dependency * [autofix.ci] apply automated fixes * [autofix.ci] apply automated fixes (attempt 2/3) * fix: Remove response model as its automatically configured. Co-authored-by: Gabriel Luiz Freitas Almeida * [autofix.ci] apply automated fixes * feat: Implement synchronous workflow execution API * [autofix.ci] apply automated fixes * [autofix.ci] apply automated fixes (attempt 2/3) * Apply suggestion from @codeflash-ai[bot] Co-authored-by: codeflash-ai[bot] <148906541+codeflash-ai[bot]@users.noreply.github.com> * Apply suggestion from @codeflash-ai[bot] Co-authored-by: codeflash-ai[bot] <148906541+codeflash-ai[bot]@users.noreply.github.com> * Apply suggestion from @codeflash-ai[bot] Co-authored-by: codeflash-ai[bot] <148906541+codeflash-ai[bot]@users.noreply.github.com> * [autofix.ci] apply automated fixes * Clean up V2 Workflow API code and add unit tests * [autofix.ci] apply automated fixes * [autofix.ci] apply automated fixes (attempt 2/3) * fix: CodeRabbit review fixes * fix: improve V2 workflow API test quality and code structure * Fix ruff whitespace issues in v2 API test files * [autofix.ci] apply automated fixes * Fix mypy type errors in v2 API files * [autofix.ci] apply automated fixes * refactor: improve data extraction with flexible path combinations and prevent flow data mutation * fix: resolve component index merge conflict after sync * [autofix.ci] apply automated fixes * [autofix.ci] apply automated fixes (attempt 2/3) * refactor: move ComponentOutput construction to avoid dict change after construction * [autofix.ci] apply automated fixes --------- Co-authored-by: Janardan S Kavia Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: Gabriel Luiz Freitas Almeida Co-authored-by: Janardan S Kavia Co-authored-by: codeflash-ai[bot] <148906541+codeflash-ai[bot]@users.noreply.github.com> --- src/backend/base/langflow/api/v1/endpoints.py | 4 +- .../base/langflow/api/v2/converters.py | 538 +++++++ src/backend/base/langflow/api/v2/workflow.py | 423 +++++- src/backend/base/langflow/exceptions/api.py | 12 + src/backend/base/langflow/helpers/flow.py | 2 +- .../tests/unit/api/v2/test_converters.py | 1237 +++++++++++++++++ .../tests/unit/api/v2/test_workflow.py | 876 +++++++++++- src/lfx/src/lfx/schema/workflow.py | 42 +- 8 files changed, 3076 insertions(+), 58 deletions(-) create mode 100644 src/backend/base/langflow/api/v2/converters.py create mode 100644 src/backend/tests/unit/api/v2/test_converters.py diff --git a/src/backend/base/langflow/api/v1/endpoints.py b/src/backend/base/langflow/api/v1/endpoints.py index 845222674b..9e69d6ad86 100644 --- a/src/backend/base/langflow/api/v1/endpoints.py +++ b/src/backend/base/langflow/api/v1/endpoints.py @@ -488,7 +488,7 @@ async def _run_flow_internal( async def simplified_run_flow( *, background_tasks: BackgroundTasks, - flow: Annotated[FlowRead | None, Depends(get_flow_by_id_or_endpoint_name)], + flow: Annotated[FlowRead, Depends(get_flow_by_id_or_endpoint_name)], input_request: SimplifiedAPIRequest | None = None, stream: bool = False, api_key_user: Annotated[UserRead, Depends(api_key_security)], @@ -546,7 +546,7 @@ async def simplified_run_flow( async def simplified_run_flow_session( *, background_tasks: BackgroundTasks, - flow: Annotated[FlowRead | None, Depends(get_flow_by_id_or_endpoint_name)], + flow: Annotated[FlowRead, Depends(get_flow_by_id_or_endpoint_name)], input_request: SimplifiedAPIRequest | None = None, stream: bool = False, api_key_user: CurrentActiveUser, diff --git a/src/backend/base/langflow/api/v2/converters.py b/src/backend/base/langflow/api/v2/converters.py new file mode 100644 index 0000000000..d05303446e --- /dev/null +++ b/src/backend/base/langflow/api/v2/converters.py @@ -0,0 +1,538 @@ +"""Schema converters for V2 Workflow API. + +This module provides conversion functions between the new V2 workflow schemas +and the existing V1 schemas, enabling reuse of existing execution logic while +presenting a new API interface. + +Key Functions: + - parse_flat_inputs: Converts flat input format to tweaks structure + - run_response_to_workflow_response: Converts V1 RunResponse to V2 WorkflowExecutionResponse + - create_error_response: Creates standardized error responses + - create_job_response: Creates background job responses + +Internal Helpers: + - _extract_nested_value: Safely extracts nested values from dict/object structures + - _extract_text_from_message: Extracts plain text from various message formats + - _simplify_output_content: Simplifies output content based on type + - _get_raw_content: Extracts raw content from vertex output data + - _extract_model_source: Extracts model information from LLM outputs + - _extract_file_path: Extracts file path from SaveToFile outputs + - _build_metadata_for_non_output: Builds metadata for non-output terminal nodes +""" + +from __future__ import annotations + +import time +from typing import TYPE_CHECKING, Any + +from lfx.schema.workflow import ( + ComponentOutput, + ErrorDetail, + JobStatus, + WorkflowExecutionRequest, + WorkflowExecutionResponse, + WorkflowJobResponse, +) + +if TYPE_CHECKING: + from lfx.graph.graph.base import Graph + + from langflow.api.v1.schemas import RunResponse + + +def parse_flat_inputs(inputs: dict[str, Any]) -> tuple[dict[str, dict[str, Any]], str | None]: + """Parse flat inputs structure into tweaks and session_id. + + Format: {"component_id.param": value} + Example: {"ChatInput-abc.input_value": "hi", "LLM-xyz.temperature": 0.7} + + All parameters (including input_value) are treated as tweaks. + The graph's topological sort handles execution order automatically. + + Args: + inputs: The inputs dictionary from WorkflowExecutionRequest + + Returns: + Tuple of (tweaks_dict, session_id) + - tweaks_dict: {component_id: {param: value}} + - session_id: Session ID if provided + + Example: + >>> inputs = { + ... "ChatInput-abc.input_value": "hello", + ... "ChatInput-abc.session_id": "session-123", + ... "LLM-xyz.temperature": 0.7 + ... } + >>> tweaks, session_id = parse_flat_inputs(inputs) + >>> tweaks + {'ChatInput-abc': {'input_value': 'hello'}, 'LLM-xyz': {'temperature': 0.7}} + >>> session_id + 'session-123' + """ + tweaks: dict[str, dict[str, Any]] = {} + session_id: str | None = None + + # Group inputs by component_id + component_inputs: dict[str, dict[str, Any]] = {} + + for key, value in inputs.items(): + if "." in key: + # Split component_id.param + component_id, param_name = key.split(".", 1) + + if component_id not in component_inputs: + component_inputs[component_id] = {} + component_inputs[component_id][param_name] = value + # No dot - treat as component-level dict (for backward compatibility) + elif isinstance(value, dict): + tweaks[key] = value + + # Process component inputs + for component_id, params in component_inputs.items(): + # Extract session_id if present (use first one found) + if "session_id" in params and session_id is None: + session_id = params["session_id"] + + # Build tweaks for all parameters except session_id + tweak_params = {k: v for k, v in params.items() if k != "session_id"} + if tweak_params: + tweaks[component_id] = tweak_params + + return tweaks, session_id + + +def _extract_nested_value(data: Any, *keys: str) -> Any: + """Safely extract nested value from dict-like structure. + + Args: + data: The data structure to extract from + *keys: Sequence of keys to traverse + + Returns: + The extracted value or None if not found + + Example: + >>> _extract_nested_value({'a': {'b': 'value'}}, 'a', 'b') + 'value' + """ + current = data + for key in keys: + if isinstance(current, dict): + current = current.get(key) + elif hasattr(current, key): + current = getattr(current, key) + else: + return None + if current is None: + return None + return current + + +def _extract_text_from_message(content: dict) -> str | None: + """Extract plain text from nested message structures. + + Handles various message formats by trying common paths in order: + - {'message': {'message': 'text', 'type': 'text'}} + - {'text': {'message': 'text'}} + - {'message': {'text': 'text'}} + - {'message': 'text'} + - {'text': {'text': 'text'}} + - {'text': 'text'} + + Args: + content: The message content dict + + Returns: + Extracted text string or None + """ + paths = [ + ("message", "message"), + ("text", "message"), + ("message", "text"), + ("message",), + ("text", "text"), + ("text",), + ] + for path in paths: + text = _extract_nested_value(content, *path) + if isinstance(text, str): + return text + return None + + +def _extract_model_source(raw_content: dict, vertex_id: str, vertex_display_name: str) -> dict | None: + """Extract model source information from LLM component output. + + Args: + raw_content: The raw output data + vertex_id: Vertex ID + vertex_display_name: Vertex display name + + Returns: + Source info dict or None + """ + model_name = _extract_nested_value(raw_content, "model_output", "message", "model_name") + if model_name: + return {"id": vertex_id, "display_name": vertex_display_name, "source": model_name} + return None + + +def _extract_file_path(raw_content: dict, vertex_type: str) -> str | None: + """Extract file path from SaveToFile component output. + + Args: + raw_content: The raw output data + vertex_type: The vertex type + + Returns: + File path string or None + """ + if vertex_type != "SaveToFile": + return None + + # Extract the message from SaveToFile component + # Return the whole message instead of filtering by specific wording + file_msg = _extract_nested_value(raw_content, "message", "message") + if isinstance(file_msg, str): + return file_msg + + return None + + +def _get_raw_content(vertex_output_data: Any) -> Any: + """Extract raw content from vertex output data. + + Tries multiple fields in order: outputs, results, messages. + Note: Uses 'is not None' checks to avoid treating empty collections as missing. + + Args: + vertex_output_data: The output data from RunResponse + + Returns: + Raw content or None + """ + if hasattr(vertex_output_data, "outputs") and vertex_output_data.outputs is not None: + return vertex_output_data.outputs + if hasattr(vertex_output_data, "results") and vertex_output_data.results is not None: + return vertex_output_data.results + if hasattr(vertex_output_data, "messages") and vertex_output_data.messages is not None: + return vertex_output_data.messages + if isinstance(vertex_output_data, dict): + # Check for 'results' first, then 'content' if results is None + if "results" in vertex_output_data: + return vertex_output_data["results"] + if "content" in vertex_output_data: + return vertex_output_data["content"] + return vertex_output_data + + +def _simplify_output_content(content: Any, output_type: str) -> Any: + """Simplify output content for output nodes. + + For message types, extracts plain text from nested structures. + For data/dataframe types, extracts the actual data value. + For other types, returns content as-is. + + Args: + content: The raw content + output_type: The output type + + Returns: + Simplified content + """ + if not isinstance(content, dict): + return content + + if output_type in {"message", "text"}: + text = _extract_text_from_message(content) + return text if text is not None else content + + if output_type == "data": + # For data types, try multiple path combinations in order + # This allows flexibility for different component output structures + data_paths = [ + ("result", "message"), # Standard: {'result': {'message': {...}}} + ("results", "message"), # Plural variant: {'results': {'message': {...}}} + ] + for path in data_paths: + 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 = [ + ("results", "message"), # Plural: {'results': {'message': {...}}} + ("result", "message"), # Singular fallback: {'result': {'message': {...}}} + ("run_sql_query", "message"), # SQL component specific + ] + for path in dataframe_paths: + dataframe_data = _extract_nested_value(content, *path) + if dataframe_data is not None: + return dataframe_data + + return content + + +def _build_metadata_for_non_output( + raw_content: Any, vertex_id: str, vertex_display_name: str, vertex_type: str, output_type: str +) -> dict[str, Any]: + """Build metadata for non-output terminal nodes. + + Extracts: + - source: Model information for LLM components + - file_path: File path for SaveToFile components + + Args: + raw_content: The raw output data + vertex_id: Vertex ID + vertex_display_name: Vertex display name + vertex_type: Vertex type + output_type: Output type + + Returns: + Metadata dict + """ + metadata: dict[str, Any] = {} + + if output_type != "message" or not isinstance(raw_content, dict): + return metadata + + # Extract model source for LLM components + source_info = _extract_model_source(raw_content, vertex_id, vertex_display_name) + if source_info: + metadata["source"] = source_info + + # Extract file path for SaveToFile components + file_path = _extract_file_path(raw_content, vertex_type) + if file_path: + metadata["file_path"] = file_path + + return metadata + + +def _process_terminal_vertex( + vertex: Any, + output_data_map: dict[str, Any], + display_name_counts: dict[str, int], +) -> tuple[str, ComponentOutput]: + """Process a single terminal vertex and return (output_key, component_output). + + Args: + vertex: The vertex to process + output_data_map: Map of component_id to output data + display_name_counts: Count of each display_name for duplicate detection + + Returns: + Tuple of (output_key, ComponentOutput) + """ + # Get output data by vertex.id (component_id) + vertex_output_data = output_data_map.get(vertex.id) + + # Determine output type from vertex + output_type = "unknown" + if vertex.outputs and len(vertex.outputs) > 0: + types = vertex.outputs[0].get("types", []) + if types: + output_type = types[0].lower() + if output_type == "unknown" and vertex.vertex_type: + output_type = vertex.vertex_type.lower() + + # Initialize metadata with component_type + metadata: dict[str, Any] = {"component_type": vertex.vertex_type} + + # Extract content + content = None + if vertex_output_data: + raw_content = _get_raw_content(vertex_output_data) + + if vertex.is_output and raw_content is not None: + # Output nodes: simplify content + content = _simplify_output_content(raw_content, output_type) + elif not vertex.is_output and raw_content is not None: + # 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"]: + # Show data content for non-output data nodes + content = _simplify_output_content(raw_content, output_type) + else: + # For message types, extract metadata only + extra_metadata = _build_metadata_for_non_output( + raw_content, + vertex.id, + vertex.display_name or vertex.vertex_type, + vertex.vertex_type, + output_type, + ) + metadata.update(extra_metadata) + + # Add any additional metadata from result data + if hasattr(vertex_output_data, "metadata") and vertex_output_data.metadata: + metadata.update(vertex_output_data.metadata) + elif isinstance(vertex_output_data, dict) and "metadata" in vertex_output_data: + result_metadata = vertex_output_data.get("metadata") + if isinstance(result_metadata, dict): + metadata.update(result_metadata) + + # Determine output key: use display_name if unique, otherwise use id + display_name = vertex.display_name or vertex.id + if display_name_counts.get(display_name, 0) > 1: + # Duplicate display_name detected, use id instead + output_key = vertex.id + # Store the display_name in metadata for reference + if vertex.display_name and vertex.display_name != vertex.id: + metadata["display_name"] = vertex.display_name + else: + # Unique display_name, use it as key + output_key = display_name + + # Build ComponentOutput + component_output = ComponentOutput( + type=output_type, + component_id=vertex.id, + status=JobStatus.COMPLETED, + content=content, + metadata=metadata, + ) + return output_key, component_output + + +def run_response_to_workflow_response( + run_response: RunResponse, + flow_id: str, + job_id: str, + workflow_request: WorkflowExecutionRequest, + graph: Graph, +) -> WorkflowExecutionResponse: + """Convert V1 RunResponse to V2 WorkflowExecutionResponse. + + This function transforms the V1 execution response to the new V2 schema format. + It intelligently handles different node types and determines what content to expose. + + Terminal Node Processing Logic: + 1. Identifies all terminal nodes (vertices with no successors) + 2. For each terminal node: + - Output nodes (is_output=True): Full content is exposed + - Data/DataFrame nodes: Content is exposed regardless of is_output flag + - Message nodes (non-output): Only metadata is exposed (source, file_path) + + Output Key Selection: + - Uses vertex.display_name as the primary key for outputs + - Falls back to vertex.id if duplicate display_names are detected + - Stores original display_name in metadata when using id as key + + Args: + run_response: The V1 response from simple_run_flow containing execution results + flow_id: The flow identifier + job_id: The generated job ID for tracking this execution + workflow_request: Original workflow request (inputs are echoed back in response) + graph: The Graph instance used for terminal node detection and vertex metadata + + Returns: + WorkflowExecutionResponse: V2 schema response with structured outputs + + Example: + Terminal nodes: ["ChatOutput-abc", "LLM-xyz", "DataNode-123"] + - ChatOutput-abc (is_output=True, type=message): Full content exposed + - LLM-xyz (is_output=False, type=message): Only metadata (model source) + - DataNode-123 (is_output=False, type=data): Full content exposed + """ + # Get terminal nodes (vertices with no successors) + try: + terminal_node_ids = graph.get_terminal_nodes() + except AttributeError: + # Fallback: manually check successor_map + terminal_node_ids = [vertex.id for vertex in graph.vertices if not graph.successor_map.get(vertex.id, [])] + + # Build output data map from run_response using component_id as key + # This ensures unique keys even when components have duplicate display_names + output_data_map: dict[str, Any] = {} + if run_response.outputs: + for run_output in run_response.outputs: + if hasattr(run_output, "outputs") and run_output.outputs: + for result_data in run_output.outputs: + if not result_data: + continue + # Use component_id as key to ensure uniqueness + component_id = result_data.component_id if hasattr(result_data, "component_id") else None + if component_id: + output_data_map[component_id] = result_data + + # First pass: collect all terminal vertices and check for duplicate display_names + terminal_vertices = [graph.get_vertex(vertex_id) for vertex_id in terminal_node_ids] + display_name_counts: dict[str, int] = {} + for vertex in terminal_vertices: + display_name = vertex.display_name or vertex.id + display_name_counts[display_name] = display_name_counts.get(display_name, 0) + 1 + + # Process each terminal vertex + outputs: dict[str, ComponentOutput] = {} + for vertex in terminal_vertices: + output_key, component_output = _process_terminal_vertex(vertex, output_data_map, display_name_counts) + outputs[output_key] = component_output + + return WorkflowExecutionResponse( + flow_id=flow_id, + job_id=job_id, + object="response", + created_timestamp=str(int(time.time())), + status=JobStatus.COMPLETED, + errors=[], + inputs=workflow_request.inputs or {}, + outputs=outputs, + metadata={}, + ) + + +def create_job_response(job_id: str) -> WorkflowJobResponse: + """Create a background job response. + + Args: + job_id: The generated job ID + + Returns: + WorkflowJobResponse for background execution + """ + return WorkflowJobResponse( + job_id=job_id, + created_timestamp=str(int(time.time())), + status=JobStatus.QUEUED, + errors=[], + ) + + +def create_error_response( + flow_id: str, + job_id: str, + workflow_request: WorkflowExecutionRequest, + error: Exception, +) -> WorkflowExecutionResponse: + """Create an error response in workflow format. + + Args: + flow_id: The flow ID + job_id: The job ID + workflow_request: Original request + error: The exception that occurred + + Returns: + WorkflowExecutionResponse with error details + """ + error_detail = ErrorDetail( + error=str(error), code="EXECUTION_ERROR", details={"flow_id": flow_id, "error_type": type(error).__name__} + ) + + return WorkflowExecutionResponse( + flow_id=flow_id, + job_id=job_id, + object="response", + created_timestamp=str(int(time.time())), + status=JobStatus.FAILED, + errors=[error_detail], + inputs=workflow_request.inputs or {}, + outputs={}, + metadata={}, + ) diff --git a/src/backend/base/langflow/api/v2/workflow.py b/src/backend/base/langflow/api/v2/workflow.py index cca3c36e79..05eb5cc694 100644 --- a/src/backend/base/langflow/api/v2/workflow.py +++ b/src/backend/base/langflow/api/v2/workflow.py @@ -1,11 +1,34 @@ -"""V2 Workflow execution endpoints.""" +"""V2 Workflow execution endpoints. + +This module implements the V2 Workflow API endpoints for executing flows with +enhanced error handling, timeout protection, and structured responses. + +Endpoints: + POST /workflow: Execute a workflow (sync, stream, or background modes) + GET /workflow: Get workflow job status by job_id + POST /workflow/stop: Stop a running workflow execution + +Features: + - Developer API protection (requires developer_api_enabled setting) + - Comprehensive error handling with structured error responses + - Timeout protection for long-running executions + - Support for multiple execution modes (sync, stream, background) + - API key authentication required for all endpoints + +Configuration: + EXECUTION_TIMEOUT: Maximum execution time for synchronous workflows (300 seconds) +""" from __future__ import annotations +import asyncio +from copy import deepcopy from typing import Annotated +from uuid import uuid4 -from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, status +from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, Request, status from fastapi.responses import StreamingResponse +from lfx.graph.graph.base import Graph from lfx.schema.workflow import ( WORKFLOW_EXECUTION_RESPONSES, WORKFLOW_STATUS_RESPONSES, @@ -16,19 +39,48 @@ from lfx.schema.workflow import ( WorkflowStopResponse, ) from lfx.services.deps import get_settings_service +from pydantic_core import ValidationError as PydanticValidationError +from sqlalchemy.exc import OperationalError +from langflow.api.utils import extract_global_variables_from_headers +from langflow.api.v1.schemas import RunResponse +from langflow.api.v2.converters import ( + create_error_response, + parse_flat_inputs, + run_response_to_workflow_response, +) +from langflow.exceptions.api import WorkflowTimeoutError, WorkflowValidationError from langflow.helpers.flow import get_flow_by_id_or_endpoint_name +from langflow.processing.process import process_tweaks, run_graph_internal from langflow.services.auth.utils import api_key_security +from langflow.services.database.models.flow.model import FlowRead from langflow.services.database.models.user.model import UserRead +# Configuration constants +EXECUTION_TIMEOUT = 300 # 5 minutes default timeout for sync execution + def check_developer_api_enabled() -> None: - """Check if developer API is enabled, raise HTTPException if not.""" + """Check if developer API is enabled. + + This dependency function protects all workflow endpoints by verifying that + the developer API feature is enabled in the application settings. + + Raises: + HTTPException: 403 Forbidden if developer_api_enabled setting is False + + Note: + This is used as a router-level dependency to protect all workflow endpoints. + """ settings = get_settings_service().settings if not settings.developer_api_enabled: raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail="This endpoint is not available", + status_code=status.HTTP_403_FORBIDDEN, + detail={ + "error": "Developer API disabled", + "code": "DEVELOPER_API_DISABLED", + "message": "Developer API is not enabled. Contact administrator to enable this feature.", + }, ) @@ -45,21 +97,301 @@ router = APIRouter(prefix="/workflow", tags=["Workflow"], dependencies=[Depends( ) async def execute_workflow( workflow_request: WorkflowExecutionRequest, - background_tasks: BackgroundTasks, # noqa: ARG001 + background_tasks: BackgroundTasks, + http_request: Request, api_key_user: Annotated[UserRead, Depends(api_key_security)], ) -> WorkflowExecutionResponse | WorkflowJobResponse | StreamingResponse: - """Execute a workflow with multiple execution modes. + """Execute a workflow with support for multiple execution modes. - - **sync**: Returns complete results immediately (background=False, stream=False) - - **stream**: Returns server-sent events in real-time (stream=True) - - **background**: Starts job and returns job ID immediately (background=True) + This endpoint supports three execution modes: + - **Synchronous** (background=False, stream=False): Returns complete results immediately + - **Streaming** (stream=True): Returns server-sent events in real-time (not yet implemented) + - **Background** (background=True): Starts job and returns job ID (not yet implemented) + + Error Handling Strategy: + - System errors (404, 500, 503, 504): Returned as HTTP error responses + - Component execution errors: Returned as HTTP 200 with errors in response body + + Args: + workflow_request: The workflow execution request containing flow_id, inputs, and mode flags + background_tasks: FastAPI background tasks for async operations + http_request: The HTTP request object for extracting headers + api_key_user: Authenticated user from API key + + Returns: + - WorkflowExecutionResponse: For synchronous execution (HTTP 200) + - WorkflowJobResponse: For background execution (HTTP 202, not yet implemented) + - StreamingResponse: For streaming execution (not yet implemented) + + Raises: + HTTPException: + - 403: Developer API disabled + - 404: Flow not found or user lacks access + - 500: Invalid flow data or validation error + - 501: Streaming or background mode not yet implemented + - 503: Database unavailable + - 504: Execution timeout exceeded """ - flow = await get_flow_by_id_or_endpoint_name(workflow_request.flow_id, api_key_user.id) - if not flow: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Flow {workflow_request.flow_id} not found") + # Validate flow exists and user has permission + try: + flow = await get_flow_by_id_or_endpoint_name(workflow_request.flow_id, api_key_user.id) + except HTTPException as e: + # get_flow_by_id_or_endpoint_name raises HTTPException with string detail + # Convert to structured format + if e.status_code == status.HTTP_404_NOT_FOUND: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail={ + "error": "Flow not found", + "code": "FLOW_NOT_FOUND", + "message": f"Flow '{workflow_request.flow_id}' does not exist or you don't have access to it", + "flow_id": workflow_request.flow_id, + }, + ) from e + # Re-raise other HTTPExceptions as-is + raise + except PydanticValidationError as e: + # Flow data validation errors (invalid flow structure) + error_msg = f"Flow has invalid data structure: {e!s}" + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail={ + "error": "Invalid flow data", + "code": "INVALID_FLOW_DATA", + "message": error_msg, + "flow_id": workflow_request.flow_id, + }, + ) from e + except OperationalError as e: + # Database errors specifically + error_msg = f"Failed to fetch flow: {e!s}" + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail={ + "error": "Service unavailable", + "code": "DATABASE_ERROR", + "message": error_msg, + "flow_id": workflow_request.flow_id, + }, + ) from e - # TODO: Implementation - raise HTTPException(status_code=status.HTTP_501_NOT_IMPLEMENTED, detail="Not implemented /workflow execution yet") + # Generate job_id for tracking + job_id = str(uuid4()) + + # Phase 1: Background mode (to be implemented) + if workflow_request.background: + raise HTTPException( + status_code=status.HTTP_501_NOT_IMPLEMENTED, + detail={ + "error": "Not implemented", + "code": "NOT_IMPLEMENTED", + "message": "Background execution not yet implemented", + }, + ) + + # Phase 2: Streaming mode (to be implemented) + if workflow_request.stream: + raise HTTPException( + status_code=status.HTTP_501_NOT_IMPLEMENTED, + detail={ + "error": "Not implemented", + "code": "NOT_IMPLEMENTED", + "message": "Streaming execution not yet implemented", + }, + ) + + # Phase 3: Synchronous execution (default) + # Note: flow is guaranteed to be non-None here because get_flow_by_id_or_endpoint_name + # raises HTTPException if flow is not found + try: + return await execute_sync_workflow_with_timeout( + workflow_request=workflow_request, + flow=flow, + job_id=job_id, + api_key_user=api_key_user, + background_tasks=background_tasks, + http_request=http_request, + ) + except WorkflowTimeoutError: + raise HTTPException( + status_code=status.HTTP_504_GATEWAY_TIMEOUT, + detail={ + "error": "Execution timeout", + "code": "EXECUTION_TIMEOUT", + "message": f"Workflow execution exceeded {EXECUTION_TIMEOUT} seconds", + "job_id": job_id, + "flow_id": workflow_request.flow_id, + "timeout_seconds": EXECUTION_TIMEOUT, + }, + ) from None + except WorkflowValidationError as e: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail={ + "error": "Workflow validation error", + "code": "INVALID_FLOW_DATA", + "message": str(e), + "job_id": job_id, + "flow_id": workflow_request.flow_id, + }, + ) from e + + +async def execute_sync_workflow_with_timeout( + workflow_request: WorkflowExecutionRequest, + flow: FlowRead, + job_id: str, + api_key_user: UserRead, + background_tasks: BackgroundTasks, + http_request: Request, +) -> WorkflowExecutionResponse: + """Execute workflow with timeout protection. + + Args: + workflow_request: The workflow execution request + flow: The flow to execute + job_id: Generated job ID for tracking + api_key_user: Authenticated user + background_tasks: FastAPI background tasks + http_request: The HTTP request object for extracting headers + + Returns: + WorkflowExecutionResponse with complete results + + Raises: + WorkflowTimeoutError: If execution exceeds timeout + WorkflowValidationError: If flow validation fails + """ + try: + return await asyncio.wait_for( + execute_sync_workflow( + workflow_request=workflow_request, + flow=flow, + job_id=job_id, + api_key_user=api_key_user, + background_tasks=background_tasks, + http_request=http_request, + ), + timeout=EXECUTION_TIMEOUT, + ) + except asyncio.TimeoutError as e: + msg = f"Execution exceeded {EXECUTION_TIMEOUT} seconds" + raise WorkflowTimeoutError(msg) from e + + +async def execute_sync_workflow( + workflow_request: WorkflowExecutionRequest, + flow: FlowRead, + job_id: str, + api_key_user: UserRead, + background_tasks: BackgroundTasks, # noqa: ARG001 + http_request: Request, +) -> WorkflowExecutionResponse: + """Execute workflow synchronously and return complete results. + + This function implements a two-tier error handling strategy: + 1. System-level errors (validation, graph build): Raised as exceptions + 2. Component execution errors: Returned in response body with HTTP 200 + + This approach allows clients to receive partial results even when some + components fail, which is useful for debugging and incremental processing. + + Execution Flow: + 1. Parse flat inputs into tweaks and session_id + 2. Validate flow data exists + 3. Extract context from HTTP headers + 4. Build graph from flow data with tweaks applied + 5. Identify terminal nodes for execution + 6. Execute graph and collect results + 7. Convert V1 RunResponse to V2 WorkflowExecutionResponse + + Args: + workflow_request: The workflow execution request with inputs and configuration + flow: The flow model from database + job_id: Generated job ID for tracking this execution + api_key_user: Authenticated user for permission checks + background_tasks: FastAPI background tasks (unused in sync mode) + http_request: The HTTP request object for extracting headers + + Returns: + WorkflowExecutionResponse: Complete execution results with outputs and metadata + + Raises: + WorkflowValidationError: If flow data is None or graph build fails + """ + # Parse flat inputs structure + tweaks, session_id = parse_flat_inputs(workflow_request.inputs or {}) + + # Validate flow data - this is a system error, not execution error + if flow.data is None: + msg = f"Flow {flow.id} has no data. The flow may be corrupted." + raise WorkflowValidationError(msg) + + # Extract request-level variables from headers (similar to V1) + # Headers with prefix X-LANGFLOW-GLOBAL-VAR-* are extracted and made available to components + request_variables = extract_global_variables_from_headers(http_request.headers) + + # Build context from request variables (similar to V1's _run_flow_internal) + context = {"request_variables": request_variables} if request_variables else None + + # Build graph - system error if this fails + try: + flow_id_str = str(flow.id) + user_id = str(api_key_user.id) + # Use deepcopy to prevent mutation of the original flow.data + # process_tweaks modifies nested dictionaries in-place + graph_data = deepcopy(flow.data) + graph_data = process_tweaks(graph_data, tweaks, stream=False) + # Pass context to graph (similar to V1's simple_run_flow) + # This allows components to access request metadata via graph.context + graph = Graph.from_payload( + graph_data, flow_id=flow_id_str, user_id=user_id, flow_name=flow.name, context=context + ) + # Set run_id for tracing/logging (similar to V1's simple_run_flow) + graph.set_run_id(job_id) + except Exception as e: + msg = f"Failed to build graph from flow data: {e!s}" + raise WorkflowValidationError(msg) from e + + # Get terminal nodes - these are the outputs we want + terminal_node_ids = graph.get_terminal_nodes() + + # Execute graph - component errors are caught and returned in response body + try: + task_result, execution_session_id = await run_graph_internal( + graph=graph, + flow_id=flow_id_str, + session_id=session_id, + inputs=None, + outputs=terminal_node_ids, + stream=False, + ) + + # Build RunResponse + run_response = RunResponse(outputs=task_result, session_id=execution_session_id) + + # Convert to WorkflowExecutionResponse + return run_response_to_workflow_response( + run_response=run_response, + flow_id=workflow_request.flow_id, + job_id=job_id, + workflow_request=workflow_request, + graph=graph, + ) + + except asyncio.CancelledError: + # Re-raise CancelledError to allow timeout mechanism to work properly + # This ensures asyncio.wait_for() can properly cancel and raise TimeoutError + raise + except Exception as exc: # noqa: BLE001 + # Component execution errors - return in response body with HTTP 200 + # This allows partial results and detailed error information per component + return create_error_response( + flow_id=workflow_request.flow_id, + job_id=job_id, + workflow_request=workflow_request, + error=exc, + ) @router.get( @@ -74,13 +406,38 @@ async def get_workflow_status( api_key_user: Annotated[UserRead, Depends(api_key_security)], # noqa: ARG001 job_id: Annotated[str, Query(description="Job ID to query")], # noqa: ARG001 ) -> WorkflowExecutionResponse | StreamingResponse: - """Get workflow job status and results by job ID.""" - # TODO: Implementation + """Get workflow job status and results by job ID. + + This endpoint allows clients to poll for the status of background workflow executions. + + Args: + api_key_user: Authenticated user from API key + job_id: The job ID returned from a background execution request + + Returns: + - WorkflowExecutionResponse: If job is complete or failed + - StreamingResponse: If job is still running (for streaming mode) + + Raises: + HTTPException: + - 403: Developer API disabled or unauthorized + - 404: Job ID not found + - 501: Not yet implemented + + Note: + This endpoint is not yet implemented. It will be added in a future release + to support background and streaming execution modes. + """ + # TODO: Implement job status tracking and retrieval + # - Store job metadata in database or cache + # - Track execution progress and status + # - Return appropriate response based on job state raise HTTPException(status_code=status.HTTP_501_NOT_IMPLEMENTED, detail="Not implemented /status yet") @router.post( "/stop", + response_model=WorkflowStopResponse, summary="Stop Workflow", description="Stop a running workflow execution", ) @@ -90,9 +447,33 @@ async def stop_workflow( ) -> WorkflowStopResponse: """Stop a running workflow execution by job_id. - Parameters: - - job_id: The specific job ID to stop - - force: Whether to force stop the workflow (optional) + This endpoint allows clients to gracefully or forcefully stop a running workflow. + + Args: + request: Stop request containing job_id and optional force flag + api_key_user: Authenticated user from API key + + Returns: + WorkflowStopResponse: Confirmation of stop request with final job status + + Raises: + HTTPException: + - 403: Developer API disabled or unauthorized + - 404: Job ID not found + - 409: Job already completed or cannot be stopped + - 501: Not yet implemented + + Note: + This endpoint is not yet implemented. It will be added in a future release + to support graceful cancellation of background and streaming executions. + + Planned behavior: + - force=False: Graceful shutdown (complete current component) + - force=True: Immediate termination """ - # TODO: Implementation + # TODO: Implement workflow cancellation + # - Locate running job by job_id + # - Send cancellation signal to execution engine + # - Handle graceful vs forced termination + # - Update job status and return response raise HTTPException(status_code=status.HTTP_501_NOT_IMPLEMENTED, detail="Not implemented /stop yet") diff --git a/src/backend/base/langflow/exceptions/api.py b/src/backend/base/langflow/exceptions/api.py index 1997ad1577..5e7e3741ae 100644 --- a/src/backend/base/langflow/exceptions/api.py +++ b/src/backend/base/langflow/exceptions/api.py @@ -10,6 +10,18 @@ class InvalidChatInputError(Exception): pass +class WorkflowExecutionError(Exception): + """Base exception for workflow execution errors.""" + + +class WorkflowTimeoutError(WorkflowExecutionError): + """Workflow execution timeout.""" + + +class WorkflowValidationError(WorkflowExecutionError): + """Workflow validation error (e.g., invalid flow data, graph build failure).""" + + # create a pidantic documentation for this class class ExceptionBody(BaseModel): message: str | list[str] diff --git a/src/backend/base/langflow/helpers/flow.py b/src/backend/base/langflow/helpers/flow.py index fae1d1e390..a037952cc5 100644 --- a/src/backend/base/langflow/helpers/flow.py +++ b/src/backend/base/langflow/helpers/flow.py @@ -396,7 +396,7 @@ def get_arg_names(inputs: list[Vertex]) -> list[dict[str, str]]: ] -async def get_flow_by_id_or_endpoint_name(flow_id_or_name: str, user_id: str | UUID | None = None) -> FlowRead | None: +async def get_flow_by_id_or_endpoint_name(flow_id_or_name: str, user_id: str | UUID | None = None) -> FlowRead: async with session_scope() as session: endpoint_name = None try: diff --git a/src/backend/tests/unit/api/v2/test_converters.py b/src/backend/tests/unit/api/v2/test_converters.py new file mode 100644 index 0000000000..6fb0bd9751 --- /dev/null +++ b/src/backend/tests/unit/api/v2/test_converters.py @@ -0,0 +1,1237 @@ +"""Comprehensive unit tests for V2 Workflow API converters. + +This test module provides extensive coverage of the converter functions that +transform between V2 workflow schemas and V1 schemas. Tests include: + +Test Coverage: + - Input parsing and transformation (parse_flat_inputs) + - Nested value extraction from various data structures + - Text extraction from different message formats + - Model source and file path extraction + - Output content simplification + - Metadata building for non-output nodes + - Response creation (job, error, workflow responses) + - End-to-end conversion from RunResponse to WorkflowExecutionResponse + +Test Strategy: + - Uses realistic payload structures from actual components + - Covers edge cases, error conditions, and malformed data + - Tests with mock objects to simulate component outputs + - Validates proper handling of duplicate names and missing data +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import Mock + +import pytest +from langflow.api.v2.converters import ( + _build_metadata_for_non_output, + _extract_file_path, + _extract_model_source, + _extract_nested_value, + _extract_text_from_message, + _get_raw_content, + _simplify_output_content, + create_error_response, + create_job_response, + parse_flat_inputs, + run_response_to_workflow_response, +) +from lfx.schema.workflow import ( + ErrorDetail, + JobStatus, + WorkflowExecutionRequest, + WorkflowExecutionResponse, + WorkflowJobResponse, +) + + +def _setup_graph_get_vertex(graph: Mock, vertices: list[Mock]) -> None: + """Helper to setup graph.get_vertex() mock for tests. + + Args: + graph: The mock graph object + vertices: List of mock vertex objects + """ + vertex_map = {v.id: v for v in vertices} + graph.get_vertex = Mock(side_effect=lambda vid: vertex_map.get(vid)) + + +class TestParseFlatInputs: + """Test suite for parse_flat_inputs function.""" + + def test_parse_flat_inputs_basic(self): + """Test basic flat input parsing with component_id.param format.""" + inputs = { + "ChatInput-abc.input_value": "hello", + "LLM-xyz.temperature": 0.7, + } + tweaks, session_id = parse_flat_inputs(inputs) + + assert tweaks == { + "ChatInput-abc": {"input_value": "hello"}, + "LLM-xyz": {"temperature": 0.7}, + } + assert session_id is None + + def test_parse_flat_inputs_with_session_id(self): + """Test parsing with session_id extraction.""" + inputs = { + "ChatInput-abc.input_value": "hello", + "ChatInput-abc.session_id": "session-123", + "LLM-xyz.temperature": 0.7, + } + tweaks, session_id = parse_flat_inputs(inputs) + + assert tweaks == { + "ChatInput-abc": {"input_value": "hello"}, + "LLM-xyz": {"temperature": 0.7}, + } + assert session_id == "session-123" + + def test_parse_flat_inputs_multiple_session_ids(self): + """Test that first session_id is used when multiple are provided.""" + inputs = { + "ChatInput-abc.session_id": "session-first", + "ChatInput-xyz.session_id": "session-second", + } + tweaks, session_id = parse_flat_inputs(inputs) + + assert session_id == "session-first" + assert tweaks == {} + + def test_parse_flat_inputs_dict_values(self): + """Test backward compatibility with dict values (no dot notation).""" + inputs = { + "ChatInput-abc": {"input_value": "hello", "temperature": 0.5}, + } + tweaks, session_id = parse_flat_inputs(inputs) + + assert tweaks == {"ChatInput-abc": {"input_value": "hello", "temperature": 0.5}} + assert session_id is None + + def test_parse_flat_inputs_mixed_formats(self): + """Test mixed flat and dict formats.""" + inputs = { + "ChatInput-abc.input_value": "hello", + "LLM-xyz": {"temperature": 0.7, "max_tokens": 100}, + } + tweaks, session_id = parse_flat_inputs(inputs) + + assert tweaks == { + "ChatInput-abc": {"input_value": "hello"}, + "LLM-xyz": {"temperature": 0.7, "max_tokens": 100}, + } + assert session_id is None + + def test_parse_flat_inputs_empty(self): + """Test with empty inputs.""" + tweaks, session_id = parse_flat_inputs({}) + + assert tweaks == {} + assert session_id is None + + def test_parse_flat_inputs_multiple_params_same_component(self): + """Test multiple parameters for the same component.""" + inputs = { + "LLM-xyz.temperature": 0.7, + "LLM-xyz.max_tokens": 100, + "LLM-xyz.top_p": 0.9, + } + tweaks, session_id = parse_flat_inputs(inputs) + + assert tweaks == { + "LLM-xyz": { + "temperature": 0.7, + "max_tokens": 100, + "top_p": 0.9, + } + } + assert session_id is None + + def test_parse_flat_inputs_malformed_key_no_dot(self): + """Test handling of non-dict value without dot notation (edge case).""" + # Non-dict values without dots are ignored (not valid component.param format) + inputs = { + "ChatInput-abc.input_value": "hello", + "invalid_key_no_dot": "should be ignored", + "LLM-xyz.temperature": 0.7, + } + tweaks, session_id = parse_flat_inputs(inputs) + + # Only valid dot-notation keys should be parsed + assert tweaks == { + "ChatInput-abc": {"input_value": "hello"}, + "LLM-xyz": {"temperature": 0.7}, + } + assert session_id is None + + def test_parse_flat_inputs_null_values(self): + """Test handling of None/null values in inputs.""" + inputs = { + "ChatInput-abc.input_value": None, + "LLM-xyz.temperature": 0.7, + "DataNode-123.data": None, + } + tweaks, session_id = parse_flat_inputs(inputs) + + # None values should be preserved + assert tweaks == { + "ChatInput-abc": {"input_value": None}, + "LLM-xyz": {"temperature": 0.7}, + "DataNode-123": {"data": None}, + } + assert session_id is None + + def test_parse_flat_inputs_deeply_nested_dict(self): + """Test handling of deeply nested dict structures (backward compatibility).""" + inputs = {"Component-abc": {"level1": {"level2": {"level3": {"value": "deeply nested"}}}}} + tweaks, session_id = parse_flat_inputs(inputs) + + # Deeply nested dicts should be preserved as-is + assert tweaks == {"Component-abc": {"level1": {"level2": {"level3": {"value": "deeply nested"}}}}} + assert session_id is None + + def test_parse_flat_inputs_empty_collections(self): + """Test handling of empty lists and dicts as values.""" + inputs = { + "Component-1.list_param": [], + "Component-2.dict_param": {}, + "Component-3.string_param": "", + } + tweaks, session_id = parse_flat_inputs(inputs) + + assert tweaks == { + "Component-1": {"list_param": []}, + "Component-2": {"dict_param": {}}, + "Component-3": {"string_param": ""}, + } + assert session_id is None + + def test_parse_flat_inputs_explicit_none_values(self): + """Test explicit None checks - None should be preserved, not treated as missing.""" + inputs = { + "Component-1.optional_param": None, + "Component-2.required_param": "value", + "Component-3.another_none": None, + } + tweaks, session_id = parse_flat_inputs(inputs) + + # None values should be explicitly preserved + assert tweaks == { + "Component-1": {"optional_param": None}, + "Component-2": {"required_param": "value"}, + "Component-3": {"another_none": None}, + } + assert session_id is None + + def test_parse_flat_inputs_special_characters_in_keys(self): + """Test handling of special characters (dashes, underscores) in component IDs and parameter names.""" + inputs = { + "Component-with-dashes.param_with_underscores": "value1", + "Component_123.param-456": "value2", + "Component_With_Underscores.param_name": "value3", + } + tweaks, session_id = parse_flat_inputs(inputs) + + assert tweaks == { + "Component-with-dashes": {"param_with_underscores": "value1"}, + "Component_123": {"param-456": "value2"}, + "Component_With_Underscores": {"param_name": "value3"}, + } + assert session_id is None + + +class TestExtractNestedValue: + """Test suite for _extract_nested_value helper function with realistic payload structures.""" + + def test_extract_model_output_message_model_name(self): + """Test extracting model_name from LLM output structure.""" + # Realistic structure from OpenAI/Anthropic LLM components + data = { + "model_output": { + "message": { + "text": "AI response here", + "model_name": "gpt-4", + "sender": "AI", + "sender_name": "AI", + } + } + } + result = _extract_nested_value(data, "model_output", "message", "model_name") + assert result == "gpt-4" + + def test_extract_result_message_from_data_output(self): + """Test extracting result from Data component output.""" + # Structure from Data/Calculator components + data = {"result": {"message": {"result": "42"}, "type": "object"}} + result = _extract_nested_value(data, "result", "message") + assert result == {"result": "42"} + + def test_extract_message_from_chat_output(self): + """Test extracting message from ChatOutput structure.""" + # ChatOutput component structure + data = { + "message": { + "message": "Hello, how can I help you?", + "type": "text", + "sender": "AI", + } + } + result = _extract_nested_value(data, "message", "message") + assert result == "Hello, how can I help you?" + + def test_extract_nested_value_error_handling(self): + """Test error handling for missing keys and None values in path.""" + # Missing key + data = {"outputs": {"result": "value"}} + result = _extract_nested_value(data, "outputs", "nonexistent") + assert result is None + + # None in path + data = {"results": None} + result = _extract_nested_value(data, "results", "data") + assert result is None + + def test_extract_from_result_data_object(self): + """Test extracting from ResultData object with attributes.""" + # Simulating ResultData from lfx.graph.schema + obj = Mock() + obj.outputs = {"message": {"text": "output text"}} + obj.results = {"data": "result data"} + + result = _extract_nested_value(obj, "outputs", "message", "text") + assert result == "output text" + + def test_extract_text_from_output_value(self): + """Test extracting from OutputValue structure.""" + # OutputValue structure from lfx.schema.schema + data = {"message": {"text": "Hello World"}, "type": "message"} + result = _extract_nested_value(data, "message", "text") + assert result == "Hello World" + + def test_extract_from_pinecone_output(self): + """Test extracting from Pinecone vector store output structure.""" + # Pinecone vector store typical output + data = { + "results": { + "matches": [ + {"id": "vec1", "score": 0.95, "metadata": {"text": "result 1"}}, + {"id": "vec2", "score": 0.87, "metadata": {"text": "result 2"}}, + ] + } + } + result = _extract_nested_value(data, "results", "matches") + assert result is not None + assert len(result) == 2 + assert result[0]["score"] == 0.95 + + def test_extract_from_chroma_output(self): + """Test extracting from Chroma vector store output structure.""" + # Chroma vector store typical output + data = { + "results": { + "ids": [["id1", "id2"]], + "distances": [[0.1, 0.3]], + "documents": [["doc1 text", "doc2 text"]], + } + } + result = _extract_nested_value(data, "results", "documents") + assert result == [["doc1 text", "doc2 text"]] + + def test_extract_from_weaviate_output(self): + """Test extracting from Weaviate vector store output structure.""" + # Weaviate vector store typical output + data = { + "data": { + "Get": { + "Document": [ + {"text": "document 1", "_additional": {"distance": 0.15}}, + {"text": "document 2", "_additional": {"distance": 0.22}}, + ] + } + } + } + result = _extract_nested_value(data, "data", "Get", "Document") + assert result is not None + assert len(result) == 2 + assert result[0]["text"] == "document 1" + + def test_extract_from_retriever_output(self): + """Test extracting from generic retriever output structure.""" + # Generic retriever output with documents + data = { + "documents": [ + {"page_content": "Retrieved doc 1", "metadata": {"source": "file1.txt"}}, + {"page_content": "Retrieved doc 2", "metadata": {"source": "file2.txt"}}, + ] + } + result = _extract_nested_value(data, "documents") + assert result is not None + assert len(result) == 2 + assert result[0]["page_content"] == "Retrieved doc 1" + + +class TestExtractTextFromMessage: + """Test suite for _extract_text_from_message function with realistic message structures.""" + + def test_extract_from_chat_output_nested_message(self): + """Test extracting from ChatOutput component with nested message.message structure.""" + # Typical ChatOutput structure + content = { + "message": { + "message": "Hello, how can I help you today?", + "type": "text", + "sender": "AI", + "sender_name": "AI", + } + } + result = _extract_text_from_message(content) + assert result == "Hello, how can I help you today?" + + def test_extract_from_llm_output_message_text(self): + """Test extracting from LLM output with message.text structure.""" + # LLM component output structure + content = { + "message": { + "text": "This is the AI response", + "model_name": "gpt-4", + "sender": "AI", + } + } + result = _extract_text_from_message(content) + assert result == "This is the AI response" + + def test_extract_direct_message_string(self): + """Test extracting direct message string.""" + # Simple message structure + content = {"message": "Direct message text"} + result = _extract_text_from_message(content) + assert result == "Direct message text" + + def test_extract_from_text_message_structure(self): + """Test extracting from text.message structure (rare but possible).""" + # Alternative structure where text contains message + content = {"text": {"message": "Text contains message"}} + result = _extract_text_from_message(content) + assert result == "Text contains message" + + def test_extract_from_text_text_structure(self): + """Test extracting from text.text nested structure.""" + # Nested text structure + content = {"text": {"text": "Nested text value"}} + result = _extract_text_from_message(content) + assert result == "Nested text value" + + def test_extract_direct_text_string(self): + """Test extracting direct text string.""" + # Simple text structure + content = {"text": "Direct text value"} + result = _extract_text_from_message(content) + assert result == "Direct text value" + + def test_extract_priority_message_message_first(self): + """Test that message.message takes priority over other fields.""" + content = { + "message": {"message": "Priority Message", "text": "Should not return this"}, + "text": "Also should not return this", + } + result = _extract_text_from_message(content) + assert result == "Priority Message" + + def test_extract_priority_message_text_over_direct_text(self): + """Test that message.text is checked before direct text.""" + content = { + "message": {"text": "Message Text"}, + "text": "Direct Text", + } + result = _extract_text_from_message(content) + assert result == "Message Text" + + def test_extract_from_output_value_message_structure(self): + """Test extracting from OutputValue message structure.""" + # OutputValue from lfx.schema.schema + content = { + "message": { + "message": "Output value message", + "type": "message", + }, + "type": "message", + } + result = _extract_text_from_message(content) + assert result == "Output value message" + + def test_extract_no_extractable_text(self): + """Test when no text can be extracted from various structures.""" + # No text fields present + content = {"data": "some data", "type": "object", "results": {}} + result = _extract_text_from_message(content) + assert result is None + + # Empty dict + content = {} + result = _extract_text_from_message(content) + assert result is None + + def test_extract_non_string_values(self): + """Test with non-string values in message/text fields.""" + content = {"message": {"message": 123, "text": ["list", "of", "items"]}} + result = _extract_text_from_message(content) + assert result is None + + def test_extract_text_circular_reference(self): + """Test handling of circular references (should not cause infinite loop).""" + # Create a circular reference structure + content: dict[str, Any] = {"message": {}} + content["message"]["self_ref"] = content # Circular reference + content["message"]["text"] = "Should extract this" + + # Should handle gracefully and extract the text + result = _extract_text_from_message(content) + assert result == "Should extract this" + + def test_extract_text_extremely_nested(self): + """Test handling of extremely nested structures (10+ levels).""" + # Build a 12-level deep nested structure + content: dict[str, Any] = { + "level1": { + "level2": { + "level3": { + "level4": { + "level5": { + "level6": { + "level7": { + "level8": {"level9": {"level10": {"level11": {"level12": "deep value"}}}} + } + } + } + } + } + } + } + } + + # Should return None as it doesn't match expected patterns + result = _extract_text_from_message(content) + assert result is None + + def test_extract_text_mixed_types_in_path(self): + """Test handling of mixed types (list, dict, string) in extraction path.""" + # Message contains a list instead of expected dict + content = {"message": ["item1", "item2", "item3"]} + result = _extract_text_from_message(content) + assert result is None + + # Text contains an integer + content = {"text": 12345} + result = _extract_text_from_message(content) + assert result is None + + # Message.message is a list + content = {"message": {"message": ["not", "a", "string"]}} + result = _extract_text_from_message(content) + assert result is None + + def test_extract_from_embedding_output(self): + """Test extracting from embedding component output structure.""" + # OpenAI/Cohere embeddings typically return vectors, not text + # But may have metadata with text + content = {"embeddings": [[0.1, 0.2, 0.3]], "text": "Text that was embedded"} + result = _extract_text_from_message(content) + assert result == "Text that was embedded" + + def test_extract_from_tool_output(self): + """Test extracting from tool/function call output structure.""" + # Tool output with result message + content = { + "message": { + "message": "Tool executed successfully: result data", + "tool_name": "calculator", + "tool_input": {"operation": "add", "numbers": [1, 2]}, + } + } + result = _extract_text_from_message(content) + assert result == "Tool executed successfully: result data" + + +class TestExtractModelSource: + """Test suite for _extract_model_source function.""" + + def test_extract_model_source_openai(self): + """Test extracting model source from OpenAI LLM output.""" + raw_content = { + "model_output": { + "message": { + "text": "AI response", + "model_name": "gpt-4-turbo", + "sender": "AI", + } + } + } + result = _extract_model_source(raw_content, "llm-123", "OpenAI LLM") + + assert result == { + "id": "llm-123", + "display_name": "OpenAI LLM", + "source": "gpt-4-turbo", + } + + def test_extract_model_source_anthropic(self): + """Test extracting model source from Anthropic LLM output.""" + raw_content = { + "model_output": { + "message": { + "text": "Claude response", + "model_name": "claude-3-opus-20240229", + } + } + } + result = _extract_model_source(raw_content, "claude-456", "Anthropic Claude") + + assert result["source"] == "claude-3-opus-20240229" + + def test_extract_model_source_missing_model_name(self): + """Test when model_name is missing.""" + raw_content = {"model_output": {"message": {"text": "response"}}} + result = _extract_model_source(raw_content, "llm-123", "OpenAI LLM") + assert result is None + + def test_extract_model_source_missing_structure(self): + """Test when model_output structure is missing or empty.""" + # Missing structure + raw_content = {"output": "some output"} + result = _extract_model_source(raw_content, "llm-123", "OpenAI LLM") + assert result is None + + # Empty dict + result = _extract_model_source({}, "llm-123", "OpenAI LLM") + assert result is None + + +class TestExtractFilePath: + """Test suite for _extract_file_path function.""" + + def test_extract_file_path_valid(self): + """Test extracting file path from SaveToFile component.""" + raw_content = {"message": {"message": "File saved successfully to /path/to/file.txt"}} + result = _extract_file_path(raw_content, "SaveToFile") + assert result == "File saved successfully to /path/to/file.txt" + + def test_extract_file_path_case_insensitive(self): + """Test case-insensitive 'saved successfully' check.""" + raw_content = {"message": {"message": "File SAVED SUCCESSFULLY to /path/file.txt"}} + result = _extract_file_path(raw_content, "SaveToFile") + assert result == "File SAVED SUCCESSFULLY to /path/file.txt" + + def test_extract_file_path_wrong_component_type(self): + """Test that non-SaveToFile components return None.""" + raw_content = {"message": {"message": "File saved successfully to /path/to/file.txt"}} + result = _extract_file_path(raw_content, "ChatOutput") + assert result is None + + def test_extract_file_path_missing_message(self): + """Test when message structure is missing.""" + # Missing message structure + raw_content = {"output": "some output"} + result = _extract_file_path(raw_content, "SaveToFile") + assert result is None + + # Message present - should return it regardless of content + # (Changed behavior: no longer filters by "saved successfully" keyword) + raw_content = {"message": {"message": "File processing failed"}} + result = _extract_file_path(raw_content, "SaveToFile") + assert result == "File processing failed" + + +class TestGetRawContent: + """Test suite for _get_raw_content function.""" + + def test_get_raw_content_from_outputs(self): + """Test extracting from outputs attribute (ResultData structure).""" + data = Mock() + data.outputs = {"message": {"text": "output text"}} + data.results = None + data.messages = None + + result = _get_raw_content(data) + assert result == {"message": {"text": "output text"}} + + def test_get_raw_content_from_results(self): + """Test extracting from results attribute.""" + data = Mock() + data.outputs = None + data.results = {"result": "value"} + data.messages = None + + result = _get_raw_content(data) + assert result == {"result": "value"} + + def test_get_raw_content_from_messages(self): + """Test extracting from messages attribute.""" + data = Mock() + data.outputs = None + data.results = None + data.messages = [{"text": "message"}] + + result = _get_raw_content(data) + assert result == [{"text": "message"}] + + def test_get_raw_content_from_dict_results(self): + """Test extracting from dict with results or content key.""" + # Dict with results key + data = {"results": {"result": "value"}} + result = _get_raw_content(data) + assert result == {"result": "value"} + + # Dict with content key + data = {"content": {"result": "value"}} + result = _get_raw_content(data) + assert result == {"result": "value"} + + def test_get_raw_content_priority_outputs(self): + """Test that outputs takes priority over results.""" + data = Mock() + data.outputs = {"from": "outputs"} + data.results = {"from": "results"} + data.messages = None + + result = _get_raw_content(data) + assert result == {"from": "outputs"} + + def test_get_raw_content_fallback(self): + """Test fallback returns data as-is.""" + data = "raw string data" + result = _get_raw_content(data) + assert result == "raw string data" + + +class TestSimplifyOutputContent: + """Test suite for _simplify_output_content function.""" + + def test_simplify_message_type(self): + """Test simplifying message type content.""" + content = {"message": {"message": "Hello World"}} + result = _simplify_output_content(content, "message") + assert result == "Hello World" + + def test_simplify_text_type(self): + """Test simplifying text type content.""" + content = {"text": "Hello World"} + result = _simplify_output_content(content, "text") + assert result == "Hello World" + + def test_simplify_data_type(self): + """Test simplifying data type content.""" + content = {"result": {"message": {"result": "4"}, "type": "object"}} + result = _simplify_output_content(content, "data") + assert result == {"result": "4"} + + def test_simplify_data_type_no_extraction(self): + """Test data type when extraction path doesn't exist.""" + content = {"data": "raw data"} + result = _simplify_output_content(content, "data") + assert result == {"data": "raw data"} + + def test_simplify_unknown_type(self): + """Test that unknown types return content as-is.""" + content = {"custom": "data"} + result = _simplify_output_content(content, "custom_type") + assert result == {"custom": "data"} + + def test_simplify_non_dict_content(self): + """Test that non-dict content is returned as-is.""" + content = "plain string" + result = _simplify_output_content(content, "message") + assert result == "plain string" + + def test_simplify_message_no_text_found(self): + """Test message type when no text can be extracted.""" + content = {"data": "some data"} + result = _simplify_output_content(content, "message") + assert result == {"data": "some data"} + + +class TestBuildMetadataForNonOutput: + """Test suite for _build_metadata_for_non_output function.""" + + def test_build_metadata_llm_component(self): + """Test building metadata for LLM component.""" + raw_content = {"model_output": {"message": {"model_name": "gpt-4", "text": "response"}}} + metadata = _build_metadata_for_non_output(raw_content, "llm-123", "OpenAI LLM", "OpenAIModel", "message") + + assert "source" in metadata + assert metadata["source"]["source"] == "gpt-4" + assert metadata["source"]["id"] == "llm-123" + + def test_build_metadata_save_to_file(self): + """Test building metadata for SaveToFile component.""" + raw_content = {"message": {"message": "File saved successfully to /path/to/file.txt"}} + metadata = _build_metadata_for_non_output(raw_content, "save-123", "Save File", "SaveToFile", "message") + assert "file_path" in metadata + assert metadata["file_path"] == "File saved successfully to /path/to/file.txt" + + def test_build_metadata_vector_store(self): + """Test building metadata for vector store components.""" + # Pinecone vector store with index info + raw_content = { + "message": { + "message": "Stored 5 vectors in index 'documents'", + "index_name": "documents", + "dimension": 1536, + "metric": "cosine", + } + } + metadata = _build_metadata_for_non_output( + raw_content, "pinecone-123", "Pinecone Store", "PineconeVectorStore", "message" + ) + + # Should not extract special metadata (no model_name or file path) + # But the raw message structure is preserved + assert metadata == {} + + def test_build_metadata_retriever(self): + """Test building metadata for retriever components.""" + # Retriever with search metadata + raw_content = { + "message": {"message": "Retrieved 3 documents", "query": "search term", "top_k": 3, "avg_score": 0.85} + } + metadata = _build_metadata_for_non_output( + raw_content, "retriever-123", "Document Retriever", "VectorStoreRetriever", "message" + ) + + # Should not extract special metadata (no model_name or file path) + assert metadata == {} + + def test_build_metadata_both_source_and_file(self): + """Test building metadata with both source and file_path.""" + raw_content = { + "model_output": {"message": {"model_name": "gpt-4"}}, + "message": {"message": "File saved successfully to /path/file.txt"}, + } + metadata = _build_metadata_for_non_output(raw_content, "save-123", "Save File", "SaveToFile", "message") + + assert "source" in metadata + assert "file_path" in metadata + + def test_build_metadata_non_message_type(self): + """Test that non-message types return empty metadata.""" + raw_content = {"data": "some data"} + metadata = _build_metadata_for_non_output(raw_content, "comp-123", "Component", "DataProcessor", "data") + assert metadata == {} + + def test_build_metadata_non_dict_content(self): + """Test that non-dict or empty content returns empty metadata.""" + # Non-dict content + metadata = _build_metadata_for_non_output("string content", "comp-123", "Component", "TextProcessor", "message") + assert metadata == {} + + # Empty dict + metadata = _build_metadata_for_non_output({}, "comp-123", "Component", "Processor", "message") + assert metadata == {} + + +class TestCreateJobResponse: + """Test suite for create_job_response function.""" + + def test_create_job_response_structure(self): + """Test job response structure and timestamp format.""" + job_id = "job-12345" + response = create_job_response(job_id) + + assert isinstance(response, WorkflowJobResponse) + assert response.job_id == job_id + assert response.status == JobStatus.QUEUED + assert response.errors == [] + assert response.created_timestamp is not None + # Verify timestamp format + assert isinstance(response.created_timestamp, str) + assert response.created_timestamp.isdigit() + + +class TestCreateErrorResponse: + """Test suite for create_error_response function.""" + + def test_create_error_response_structure(self): + """Test error response structure.""" + flow_id = "flow-123" + job_id = "job-456" + request = WorkflowExecutionRequest(flow_id=flow_id, inputs={"test": "input"}) + error = ValueError("Test error message") + + response = create_error_response(flow_id, job_id, request, error) + + assert isinstance(response, WorkflowExecutionResponse) + assert response.flow_id == flow_id + assert response.job_id == job_id + assert response.status == JobStatus.FAILED + assert len(response.errors) == 1 + assert response.outputs == {} + + def test_create_error_response_error_details(self): + """Test error details in response.""" + error = RuntimeError("Runtime error occurred") + response = create_error_response( + "flow-1", "job-1", WorkflowExecutionRequest(flow_id="flow-1", inputs={}), error + ) + + error_detail = response.errors[0] + assert isinstance(error_detail, ErrorDetail) + assert error_detail.error == "Runtime error occurred" + assert error_detail.code == "EXECUTION_ERROR" + assert error_detail.details["error_type"] == "RuntimeError" + assert error_detail.details["flow_id"] == "flow-1" + + def test_create_error_response_preserves_inputs(self): + """Test that original inputs are preserved in error response.""" + inputs = {"component.param": "value"} + request = WorkflowExecutionRequest(flow_id="flow-1", inputs=inputs) + error = Exception("Error") + + response = create_error_response("flow-1", "job-1", request, error) + assert response.inputs == inputs + + +class TestRunResponseToWorkflowResponse: + """Test suite for run_response_to_workflow_response function.""" + + def test_run_response_basic_output_node(self): + """Test conversion with basic output node.""" + # Create mock graph + graph = Mock() + vertex = Mock() + vertex.id = "output-123" + vertex.display_name = "ChatOutput" + vertex.vertex_type = "ChatOutput" + vertex.is_output = True + vertex.outputs = [{"types": ["Message"]}] + + graph.vertices = [vertex] + graph.get_terminal_nodes = Mock(return_value=["output-123"]) + _setup_graph_get_vertex(graph, [vertex]) + + # Create mock run response + run_response = Mock() + result_data = Mock() + result_data.component_id = "output-123" + result_data.outputs = {"message": {"message": "Hello World"}} + result_data.metadata = {} + + run_output = Mock() + run_output.outputs = [result_data] + run_response.outputs = [run_output] + + # Create request + request = WorkflowExecutionRequest(flow_id="flow-123", inputs={"test": "input"}) + + # Convert + response = run_response_to_workflow_response(run_response, "flow-123", "job-456", request, graph) + + assert isinstance(response, WorkflowExecutionResponse) + assert response.flow_id == "flow-123" + assert response.job_id == "job-456" + assert response.status == JobStatus.COMPLETED + assert "ChatOutput" in response.outputs + assert response.outputs["ChatOutput"].content == "Hello World" + + def test_run_response_non_output_terminal_node(self): + """Test conversion with non-output terminal node.""" + # Create mock graph + graph = Mock() + vertex = Mock() + vertex.id = "llm-123" + vertex.display_name = "LLM" + vertex.vertex_type = "OpenAIModel" + vertex.is_output = False + vertex.outputs = [{"types": ["Message"]}] + + graph.vertices = [vertex] + graph.get_terminal_nodes = Mock(return_value=["llm-123"]) + _setup_graph_get_vertex(graph, [vertex]) + + # Create mock run response with model info + run_response = Mock() + result_data = Mock() + result_data.component_id = "llm-123" + result_data.outputs = {"model_output": {"message": {"model_name": "gpt-4", "text": "response"}}} + result_data.metadata = {} + + run_output = Mock() + run_output.outputs = [result_data] + run_response.outputs = [run_output] + + request = WorkflowExecutionRequest(flow_id="flow-1", inputs={}) + + response = run_response_to_workflow_response(run_response, "flow-1", "job-1", request, graph) + + assert "LLM" in response.outputs + output = response.outputs["LLM"] + assert output.content is None # Non-output message nodes don't show content + assert "source" in output.metadata + assert output.metadata["source"]["source"] == "gpt-4" + + def test_run_response_duplicate_display_names(self): + """Test handling of duplicate display names.""" + # Create mock graph with duplicate display names + graph = Mock() + vertex1 = Mock() + vertex1.id = "output-1" + vertex1.display_name = "Output" + vertex1.vertex_type = "ChatOutput" + vertex1.is_output = True + vertex1.outputs = [{"types": ["Message"]}] + + vertex2 = Mock() + vertex2.id = "output-2" + vertex2.display_name = "Output" + vertex2.vertex_type = "ChatOutput" + vertex2.is_output = True + vertex2.outputs = [{"types": ["Message"]}] + + graph.vertices = [vertex1, vertex2] + graph.get_terminal_nodes = Mock(return_value=["output-1", "output-2"]) + _setup_graph_get_vertex(graph, [vertex1, vertex2]) + + run_response = Mock() + run_response.outputs = [] + + request = WorkflowExecutionRequest(flow_id="flow-1", inputs={}) + + response = run_response_to_workflow_response(run_response, "flow-1", "job-1", request, graph) + + # Should use IDs instead of duplicate display names + assert "output-1" in response.outputs + assert "output-2" in response.outputs + # When duplicate display names are detected, IDs are used as keys + # The metadata contains component_type but display_name is not added in current implementation + assert response.outputs["output-1"].metadata.get("component_type") == "ChatOutput" + assert response.outputs["output-2"].metadata.get("component_type") == "ChatOutput" + + def test_run_response_data_type_non_output(self): + """Test that data type non-output nodes show content.""" + graph = Mock() + vertex = Mock() + vertex.id = "data-123" + vertex.display_name = "DataNode" + vertex.vertex_type = "DataProcessor" + vertex.is_output = False + vertex.outputs = [{"types": ["Data"]}] + + graph.vertices = [vertex] + graph.get_terminal_nodes = Mock(return_value=["data-123"]) + _setup_graph_get_vertex(graph, [vertex]) + + run_response = Mock() + result_data = Mock() + result_data.component_id = "data-123" + result_data.outputs = {"result": {"message": {"result": "42"}}} + result_data.metadata = {} + + run_output = Mock() + run_output.outputs = [result_data] + run_response.outputs = [run_output] + + request = WorkflowExecutionRequest(flow_id="flow-1", inputs={}) + + response = run_response_to_workflow_response(run_response, "flow-1", "job-1", request, graph) + + # Data type non-output nodes should show content + assert response.outputs["DataNode"].content == {"result": "42"} + + def test_run_response_fallback_terminal_detection(self): + """Test fallback terminal node detection when get_terminal_nodes fails.""" + graph = Mock() + vertex = Mock() + vertex.id = "output-123" + vertex.display_name = "Output" + vertex.vertex_type = "ChatOutput" + vertex.is_output = True + vertex.outputs = [{"types": ["Message"]}] + + graph.vertices = [vertex] + # Simulate AttributeError + graph.get_terminal_nodes = Mock(side_effect=AttributeError) + graph.successor_map = {"output-123": []} # No successors = terminal + _setup_graph_get_vertex(graph, [vertex]) + + run_response = Mock() + run_response.outputs = [] + + request = WorkflowExecutionRequest(flow_id="flow-1", inputs={}) + + response = run_response_to_workflow_response(run_response, "flow-1", "job-1", request, graph) + + assert "Output" in response.outputs + + def test_run_response_preserves_inputs(self): + """Test that inputs are preserved in response.""" + graph = Mock() + graph.vertices = [] + graph.get_terminal_nodes = Mock(return_value=[]) + _setup_graph_get_vertex(graph, []) + + run_response = Mock() + run_response.outputs = [] + + inputs = {"component.param": "value"} + request = WorkflowExecutionRequest(flow_id="flow-1", inputs=inputs) + + response = run_response_to_workflow_response(run_response, "flow-1", "job-1", request, graph) + assert response.inputs == inputs + + def test_run_response_vector_store_terminal(self): + """Test vector store as terminal node.""" + graph = Mock() + vertex = Mock() + vertex.id = "pinecone-123" + vertex.display_name = "Vector Store" + vertex.vertex_type = "PineconeVectorStore" + vertex.is_output = False + vertex.outputs = [{"types": ["Data"]}] + + graph.vertices = [vertex] + graph.get_terminal_nodes = Mock(return_value=["pinecone-123"]) + _setup_graph_get_vertex(graph, [vertex]) + + run_response = Mock() + result_data = Mock() + result_data.component_id = "pinecone-123" + result_data.outputs = {"result": {"message": {"result": {"ids": ["vec1", "vec2"], "stored_count": 2}}}} + result_data.metadata = {"index_name": "documents"} + + run_output = Mock() + run_output.outputs = [result_data] + run_response.outputs = [run_output] + + request = WorkflowExecutionRequest(flow_id="flow-1", inputs={}) + + response = run_response_to_workflow_response(run_response, "flow-1", "job-1", request, graph) + + # Data type non-output nodes should show content + assert "Vector Store" in response.outputs + assert response.outputs["Vector Store"].content is not None + assert "stored_count" in str(response.outputs["Vector Store"].content) + + def test_run_response_retriever_with_metadata(self): + """Test retriever with search metadata.""" + graph = Mock() + vertex = Mock() + vertex.id = "retriever-456" + vertex.display_name = "Retriever" + vertex.vertex_type = "VectorStoreRetriever" + vertex.is_output = False + vertex.outputs = [{"types": ["Data"]}] + + graph.vertices = [vertex] + graph.get_terminal_nodes = Mock(return_value=["retriever-456"]) + _setup_graph_get_vertex(graph, [vertex]) + + run_response = Mock() + result_data = Mock() + result_data.component_id = "retriever-456" + result_data.outputs = { + "result": {"message": {"result": {"documents": ["doc1", "doc2", "doc3"], "scores": [0.95, 0.87, 0.82]}}} + } + result_data.metadata = {"query": "search term", "top_k": 3} + + run_output = Mock() + run_output.outputs = [result_data] + run_response.outputs = [run_output] + + request = WorkflowExecutionRequest(flow_id="flow-1", inputs={}) + + response = run_response_to_workflow_response(run_response, "flow-1", "job-1", request, graph) + + # Should include content and metadata + assert "Retriever" in response.outputs + output = response.outputs["Retriever"] + assert output.content is not None + assert "documents" in str(output.content) + # Metadata from result_data should be included + assert output.metadata is not None + assert output.metadata.get("query") == "search term" + assert output.metadata.get("top_k") == 3 + + def test_run_response_empty_outputs(self): + """Test handling of empty outputs.""" + graph = Mock() + graph.vertices = [] + graph.get_terminal_nodes = Mock(return_value=[]) + _setup_graph_get_vertex(graph, []) + + run_response = Mock() + run_response.outputs = None + + request = WorkflowExecutionRequest(flow_id="flow-1", inputs={}) + + response = run_response_to_workflow_response(run_response, "flow-1", "job-1", request, graph) + + assert response.outputs == {} + assert response.status == JobStatus.COMPLETED + + def test_run_response_corrupted_vertex_data(self): + """Test handling of corrupted/malformed vertex data.""" + graph = Mock() + + # Create vertex with missing/corrupted attributes + vertex = Mock() + vertex.id = "corrupted-123" + vertex.display_name = None # Missing display name + vertex.vertex_type = None # Missing vertex type + vertex.is_output = True + vertex.outputs = None # Missing outputs + + graph.vertices = [vertex] + graph.get_terminal_nodes = Mock(return_value=["corrupted-123"]) + _setup_graph_get_vertex(graph, [vertex]) + + run_response = Mock() + run_response.outputs = [] + + request = WorkflowExecutionRequest(flow_id="flow-1", inputs={}) + + # Should handle gracefully without crashing + response = run_response_to_workflow_response(run_response, "flow-1", "job-1", request, graph) + + # Should use ID as fallback when display_name is None + assert "corrupted-123" in response.outputs + assert response.status == JobStatus.COMPLETED + + def test_run_response_missing_required_fields(self): + """Test handling when result_data is missing required fields.""" + graph = Mock() + vertex = Mock() + vertex.id = "output-123" + vertex.display_name = "Output" + vertex.vertex_type = "ChatOutput" + vertex.is_output = True + vertex.outputs = [{"types": ["Message"]}] + + graph.vertices = [vertex] + graph.get_terminal_nodes = Mock(return_value=["output-123"]) + _setup_graph_get_vertex(graph, [vertex]) + + # Create result_data without component_id + run_response = Mock() + result_data = Mock() + result_data.component_id = None # Missing component_id + result_data.outputs = {"message": "test"} + result_data.metadata = {} + + run_output = Mock() + run_output.outputs = [result_data] + run_response.outputs = [run_output] + + request = WorkflowExecutionRequest(flow_id="flow-1", inputs={}) + + # Should handle gracefully - vertex won't match result_data + response = run_response_to_workflow_response(run_response, "flow-1", "job-1", request, graph) + + # Output should exist but with no content (no matching result_data) + assert "Output" in response.outputs + assert response.outputs["Output"].content is None + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) diff --git a/src/backend/tests/unit/api/v2/test_workflow.py b/src/backend/tests/unit/api/v2/test_workflow.py index ecc7438390..e71e7569b1 100644 --- a/src/backend/tests/unit/api/v2/test_workflow.py +++ b/src/backend/tests/unit/api/v2/test_workflow.py @@ -1,10 +1,40 @@ +"""Comprehensive unit tests for V2 Workflow API endpoints. + +This test module provides extensive coverage of the workflow execution endpoints, +including authentication, authorization, error handling, and execution modes. + +Test Coverage: + - Developer API protection (enabled/disabled scenarios) + - API key authentication requirements + - Flow validation and error handling + - Database error handling + - Execution timeout protection + - Synchronous execution with various component types + - Error response structure validation + - Multiple execution modes (sync, stream, background) + +Test Organization: + - TestWorkflowDeveloperAPIProtection: Tests developer API feature flag + - TestWorkflowErrorHandling: Tests comprehensive error scenarios + - TestWorkflowSyncExecution: Tests successful execution flows + +Test Strategy: + - Uses real database with proper cleanup + - Mocks external dependencies (LLM APIs, file operations) + - Tests both success and failure paths + - Validates response structure and status codes +""" + +import asyncio from unittest.mock import MagicMock, patch from uuid import uuid4 import pytest from httpx import AsyncClient +from langflow.exceptions.api import WorkflowValidationError from langflow.services.database.models.flow.model import Flow from lfx.services.deps import session_scope +from sqlalchemy.exc import OperationalError class TestWorkflowDeveloperAPIProtection: @@ -42,9 +72,10 @@ class TestWorkflowDeveloperAPIProtection: headers=headers, ) - assert response.status_code == 404 + assert response.status_code == 403 result = response.json() - assert "This endpoint is not available" in result["detail"] + assert result["detail"]["code"] == "DEVELOPER_API_DISABLED" + assert "Developer API" in result["detail"]["message"] async def test_stop_workflow_blocked_when_dev_api_disabled( self, @@ -62,9 +93,9 @@ class TestWorkflowDeveloperAPIProtection: headers=headers, ) - assert response.status_code == 404 + assert response.status_code == 403 result = response.json() - assert "This endpoint is not available" in result["detail"] + assert result["detail"]["code"] == "DEVELOPER_API_DISABLED" @pytest.fixture def mock_settings_dev_api_enabled(self): @@ -100,9 +131,9 @@ class TestWorkflowDeveloperAPIProtection: # Should return 404 because flow doesn't exist, NOT because endpoint is disabled assert response.status_code == 404 - assert "Flow identifier" in response.text - assert "not found" in response.text - assert "This endpoint is not available" not in response.text + result = response.json() + assert result["detail"]["code"] == "FLOW_NOT_FOUND" + assert "550e8400-e29b-41d4-a716-446655440000" in result["detail"]["flow_id"] async def test_get_workflow_allowed_when_dev_api_enabled_job_not_found( self, @@ -158,9 +189,9 @@ class TestWorkflowDeveloperAPIProtection: headers=headers, ) - assert response.status_code == 404 + assert response.status_code == 403 result = response.json() - assert "This endpoint is not available" in result["detail"] + assert result["detail"]["code"] == "DEVELOPER_API_DISABLED" async def test_execute_workflow_allowed_when_dev_api_enabled_flow_exists( self, @@ -168,7 +199,7 @@ class TestWorkflowDeveloperAPIProtection: created_api_key, mock_settings_dev_api_enabled, # noqa: ARG002 ): - """Test POST /workflow allowed when dev API enabled - flow exists (501 not implemented).""" + """Test POST /workflow allowed when dev API enabled - flow exists and executes.""" flow_id = uuid4() # Create a flow in the database using the established pattern @@ -194,10 +225,17 @@ class TestWorkflowDeveloperAPIProtection: headers=headers, ) - # Should return 501 because endpoint is not implemented yet - assert response.status_code == 501 - assert "Not implemented" in response.text - assert "This endpoint is not available" not in response.text + # Should return 200 because flow is valid (empty nodes/edges is valid) + # The execution will complete successfully with no outputs + assert response.status_code == 200 + result = response.json() + + # Verify response contains expected fields with proper structure + assert "outputs" in result or "errors" in result + if "outputs" in result: + assert isinstance(result["outputs"], dict) + if "errors" in result: + assert isinstance(result["errors"], list) finally: # Clean up the flow following established pattern @@ -270,11 +308,821 @@ class TestWorkflowDeveloperAPIProtection: assert response.status_code == 403 assert "API key must be passed" in response.json()["detail"] + +class TestWorkflowErrorHandling: + """Test comprehensive error handling for workflow endpoints.""" + + @pytest.fixture + def mock_settings_dev_api_enabled(self): + """Mock settings with developer API enabled.""" + with patch("langflow.api.v2.workflow.get_settings_service") as mock_get_settings_service: + mock_service = MagicMock() + mock_settings = MagicMock() + mock_settings.developer_api_enabled = True + mock_service.settings = mock_settings + mock_get_settings_service.return_value = mock_service + yield mock_settings + + async def test_flow_not_found_returns_404_with_error_code( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """Test that non-existent flow returns 404 with FLOW_NOT_FOUND error code.""" + flow_id = str(uuid4()) + request_data = {"flow_id": flow_id, "background": False, "stream": False, "inputs": None} + + headers = {"x-api-key": created_api_key.api_key} + response = await client.post("api/v2/workflow", json=request_data, headers=headers) + + assert response.status_code == 404 + result = response.json() + assert result["detail"]["code"] == "FLOW_NOT_FOUND" + assert result["detail"]["flow_id"] == flow_id + assert "does not exist" in result["detail"]["message"] + + async def test_database_error_returns_503( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """Test that database errors return 503 with DATABASE_ERROR code.""" + flow_id = str(uuid4()) + request_data = {"flow_id": flow_id, "background": False, "stream": False, "inputs": None} + + # Mock get_flow_by_id_or_endpoint_name to raise OperationalError + with patch("langflow.api.v2.workflow.get_flow_by_id_or_endpoint_name") as mock_get_flow: + mock_get_flow.side_effect = OperationalError("statement", "params", "orig") + + headers = {"x-api-key": created_api_key.api_key} + response = await client.post("api/v2/workflow", json=request_data, headers=headers) + + assert response.status_code == 503 + result = response.json() + assert result["detail"]["code"] == "DATABASE_ERROR" + assert "Failed to fetch flow" in result["detail"]["message"] + assert result["detail"]["flow_id"] == flow_id + + async def test_flow_with_no_data_returns_500( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """Test that flow with no data returns 500 with INVALID_FLOW_DATA code.""" + flow_id = uuid4() + + # Create a flow with no data + async with session_scope() as session: + flow = Flow( + id=flow_id, + name="Flow with no data", + description="Test flow with no data", + data=None, # No data + user_id=created_api_key.user_id, + ) + session.add(flow) + await session.flush() + await session.refresh(flow) + + try: + request_data = {"flow_id": str(flow_id), "background": False, "stream": False, "inputs": None} + + headers = {"x-api-key": created_api_key.api_key} + response = await client.post("api/v2/workflow", json=request_data, headers=headers) + + assert response.status_code == 500 + result = response.json() + assert result["detail"]["code"] == "INVALID_FLOW_DATA" + assert "has no data" in result["detail"]["message"] + assert result["detail"]["flow_id"] == str(flow_id) + assert "job_id" in result["detail"] + + finally: + # Clean up + async with session_scope() as session: + flow = await session.get(Flow, flow_id) + if flow: + await session.delete(flow) + + async def test_graph_build_failure_returns_500( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """Test that graph build failure returns 500 with INVALID_FLOW_DATA code.""" + flow_id = uuid4() + + # Create a flow with invalid data that will fail validation/graph building + async with session_scope() as session: + flow = Flow( + id=flow_id, + name="Flow with invalid data", + description="Test flow with invalid graph data", + data={"invalid": "data"}, # Invalid graph data (missing 'nodes' field) + user_id=created_api_key.user_id, + ) + session.add(flow) + await session.flush() + await session.refresh(flow) + + try: + request_data = {"flow_id": str(flow_id), "background": False, "stream": False, "inputs": None} + + headers = {"x-api-key": created_api_key.api_key} + response = await client.post("api/v2/workflow", json=request_data, headers=headers) + + assert response.status_code == 500 + result = response.json() + assert result["detail"]["code"] == "INVALID_FLOW_DATA" + # The error message should indicate invalid flow data structure + error_msg = result["detail"]["message"].lower() + assert "invalid data structure" in error_msg or "must have nodes" in error_msg + assert result["detail"]["flow_id"] == str(flow_id) + + finally: + # Clean up + async with session_scope() as session: + flow = await session.get(Flow, flow_id) + if flow: + await session.delete(flow) + + async def test_execution_timeout_with_real_delay( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """Test that execution timeout works with real async delay.""" + flow_id = uuid4() + + # Create a valid flow + async with session_scope() as session: + flow = Flow( + id=flow_id, + name="Test Flow", + description="Test flow for timeout", + data={"nodes": [], "edges": []}, + user_id=created_api_key.user_id, + ) + session.add(flow) + await session.flush() + await session.refresh(flow) + + try: + request_data = {"flow_id": str(flow_id), "background": False, "stream": False, "inputs": None} + + # Mock execute_sync_workflow to sleep longer than timeout + async def slow_execution(*args, **kwargs): # noqa: ARG001 + await asyncio.sleep(2) # Sleep for 2 seconds + return MagicMock() + + # Temporarily reduce timeout for testing + with ( + patch("langflow.api.v2.workflow.execute_sync_workflow", side_effect=slow_execution), + patch("langflow.api.v2.workflow.EXECUTION_TIMEOUT", 0.5), # 0.5 second timeout + ): + headers = {"x-api-key": created_api_key.api_key} + response = await client.post("api/v2/workflow", json=request_data, headers=headers) + + assert response.status_code == 504 + result = response.json() + assert result["detail"]["code"] == "EXECUTION_TIMEOUT" + assert "exceeded" in result["detail"]["message"] + assert result["detail"]["flow_id"] == str(flow_id) + assert "job_id" in result["detail"] + + finally: + # Clean up + async with session_scope() as session: + flow = await session.get(Flow, flow_id) + if flow: + await session.delete(flow) + + async def test_background_mode_returns_501( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """Test that background mode returns 501 with NOT_IMPLEMENTED code.""" + flow_id = uuid4() + + # Create a valid flow + async with session_scope() as session: + flow = Flow( + id=flow_id, + name="Test Flow", + description="Test flow", + data={"nodes": [], "edges": []}, + user_id=created_api_key.user_id, + ) + session.add(flow) + await session.flush() + await session.refresh(flow) + + try: + request_data = { + "flow_id": str(flow_id), + "background": True, # Background mode + "stream": False, + "inputs": None, + } + + headers = {"x-api-key": created_api_key.api_key} + response = await client.post("api/v2/workflow", json=request_data, headers=headers) + + assert response.status_code == 501 + result = response.json() + assert result["detail"]["code"] == "NOT_IMPLEMENTED" + assert "Background execution not yet implemented" in result["detail"]["message"] + + finally: + # Clean up + async with session_scope() as session: + flow = await session.get(Flow, flow_id) + if flow: + await session.delete(flow) + + async def test_streaming_mode_returns_501( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """Test that streaming mode returns 501 with NOT_IMPLEMENTED code.""" + flow_id = uuid4() + + # Create a valid flow + async with session_scope() as session: + flow = Flow( + id=flow_id, + name="Test Flow", + description="Test flow", + data={"nodes": [], "edges": []}, + user_id=created_api_key.user_id, + ) + session.add(flow) + await session.flush() + await session.refresh(flow) + + try: + request_data = { + "flow_id": str(flow_id), + "background": False, + "stream": True, # Streaming mode + "inputs": None, + } + + headers = {"x-api-key": created_api_key.api_key} + response = await client.post("api/v2/workflow", json=request_data, headers=headers) + + assert response.status_code == 501 + result = response.json() + assert result["detail"]["code"] == "NOT_IMPLEMENTED" + assert "Streaming execution not yet implemented" in result["detail"]["message"] + + finally: + # Clean up + async with session_scope() as session: + flow = await session.get(Flow, flow_id) + if flow: + await session.delete(flow) + + async def test_error_response_structure( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """Test that all error responses have consistent structure.""" + flow_id = str(uuid4()) + request_data = {"flow_id": flow_id, "background": False, "stream": False, "inputs": None} + + headers = {"x-api-key": created_api_key.api_key} + response = await client.post("api/v2/workflow", json=request_data, headers=headers) + + assert response.status_code == 404 + result = response.json() + + # Verify error structure + assert "detail" in result + assert "error" in result["detail"] + assert "code" in result["detail"] + assert "message" in result["detail"] + assert "flow_id" in result["detail"] + + # Verify types + assert isinstance(result["detail"]["error"], str) + assert isinstance(result["detail"]["code"], str) + assert isinstance(result["detail"]["message"], str) + assert isinstance(result["detail"]["flow_id"], str) + + async def test_workflow_validation_error_propagation( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """Test that WorkflowValidationError is properly caught and converted to 500.""" + flow_id = uuid4() + + # Create a flow + async with session_scope() as session: + flow = Flow( + id=flow_id, + name="Test Flow", + description="Test flow", + data={"nodes": [], "edges": []}, + user_id=created_api_key.user_id, + ) + session.add(flow) + await session.flush() + await session.refresh(flow) + + try: + request_data = {"flow_id": str(flow_id), "background": False, "stream": False, "inputs": None} + + # Mock execute_sync_workflow to raise WorkflowValidationError + with patch("langflow.api.v2.workflow.execute_sync_workflow") as mock_execute: + mock_execute.side_effect = WorkflowValidationError("Test validation error") + + headers = {"x-api-key": created_api_key.api_key} + response = await client.post("api/v2/workflow", json=request_data, headers=headers) + + assert response.status_code == 500 + result = response.json() + assert result["detail"]["code"] == "INVALID_FLOW_DATA" + assert "Test validation error" in result["detail"]["message"] + + finally: + # Clean up + async with session_scope() as session: + flow = await session.get(Flow, flow_id) + if flow: + await session.delete(flow) + # Test GET /workflow without API key response = await client.get("api/v2/workflow?job_id=550e8400-e29b-41d4-a716-446655440001") assert response.status_code == 403 assert "API key must be passed" in response.json()["detail"] + +class TestWorkflowSyncExecution: + """Test synchronous workflow execution with realistic component mocking.""" + + @pytest.fixture + def mock_settings_dev_api_enabled(self): + """Mock settings with developer API enabled.""" + with patch("langflow.api.v2.workflow.get_settings_service") as mock_get_settings_service: + mock_service = MagicMock() + mock_settings = MagicMock() + mock_settings.developer_api_enabled = True + mock_service.settings = mock_settings + mock_get_settings_service.return_value = mock_service + yield mock_settings + + async def test_sync_execution_with_empty_flow_returns_200( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """Test sync execution with empty flow returns 200 with empty outputs.""" + flow_id = uuid4() + + async with session_scope() as session: + flow = Flow( + id=flow_id, + name="Empty Flow", + description="Flow with no nodes", + data={"nodes": [], "edges": []}, + user_id=created_api_key.user_id, + ) + session.add(flow) + await session.flush() + await session.refresh(flow) + + try: + request_data = {"flow_id": str(flow_id), "background": False, "stream": False, "inputs": None} + + headers = {"x-api-key": created_api_key.api_key} + response = await client.post("api/v2/workflow", json=request_data, headers=headers) + + assert response.status_code == 200 + result = response.json() + + # Verify response structure + assert "flow_id" in result + assert result["flow_id"] == str(flow_id) + assert "job_id" in result + + # Verify outputs or errors are present with actual content + assert "outputs" in result or "errors" in result + if "outputs" in result: + assert isinstance(result["outputs"], dict) + if "errors" in result: + assert isinstance(result["errors"], list) + # session_id is only present if provided in inputs + + finally: + async with session_scope() as session: + flow = await session.get(Flow, flow_id) + if flow: + await session.delete(flow) + + async def test_sync_execution_component_error_returns_200_with_error_in_body( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """Test that component execution errors return 200 with error in response body.""" + flow_id = uuid4() + + async with session_scope() as session: + flow = Flow( + id=flow_id, + name="Test Flow", + description="Flow for testing component errors", + data={"nodes": [], "edges": []}, + user_id=created_api_key.user_id, + ) + session.add(flow) + await session.flush() + await session.refresh(flow) + + try: + request_data = {"flow_id": str(flow_id), "background": False, "stream": False, "inputs": None} + + # Mock run_graph_internal to raise a component execution error + with patch("langflow.api.v2.workflow.run_graph_internal") as mock_run: + mock_run.side_effect = Exception("Component execution failed: LLM API key not configured") + + headers = {"x-api-key": created_api_key.api_key} + response = await client.post("api/v2/workflow", json=request_data, headers=headers) + + # Component errors should return 200 with error in body + assert response.status_code == 200 + result = response.json() + + # Verify error is in response body (via create_error_response) + assert "errors" in result + assert len(result["errors"]) > 0 + assert "Component execution failed" in str(result["errors"][0]) + assert result["status"] == "failed" + assert result["flow_id"] == str(flow_id) + assert "job_id" in result + + finally: + async with session_scope() as session: + flow = await session.get(Flow, flow_id) + if flow: + await session.delete(flow) + + async def test_sync_execution_with_chat_input_output( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """Test sync execution with ChatInput and ChatOutput components.""" + flow_id = uuid4() + + async with session_scope() as session: + flow = Flow( + id=flow_id, + name="Chat Flow", + description="Flow with chat input/output", + data={"nodes": [], "edges": []}, + user_id=created_api_key.user_id, + ) + session.add(flow) + await session.flush() + await session.refresh(flow) + + try: + # Input format: component_id.param = value + request_data = { + "flow_id": str(flow_id), + "background": False, + "stream": False, + "inputs": { + "ChatInput-abc123.input_value": "Hello, how are you?", + "ChatInput-abc123.session_id": "session-456", + }, + } + + # Mock successful execution with ChatOutput + mock_result_data = MagicMock() + mock_result_data.component_id = "ChatOutput-xyz789" + mock_result_data.outputs = {"message": {"message": "I'm doing well, thank you for asking!", "type": "text"}} + mock_result_data.metadata = {} + + # Wrap ResultData in RunOutputs + mock_run_output = MagicMock() + mock_run_output.outputs = [mock_result_data] + + with patch("langflow.api.v2.workflow.run_graph_internal") as mock_run: + # run_graph_internal returns tuple[list[RunOutputs], str] + mock_run.return_value = ([mock_run_output], "session-456") + + headers = {"x-api-key": created_api_key.api_key} + response = await client.post("api/v2/workflow", json=request_data, headers=headers) + + assert response.status_code == 200 + result = response.json() + + # Verify response structure + assert result["flow_id"] == str(flow_id) + assert "job_id" in result + assert "outputs" in result + # Note: Detailed content validation requires proper graph/vertex mocking + # which is beyond the scope of unit tests. Integration tests should validate content. + + # Verify inputs were echoed back + assert "inputs" in result + assert result["inputs"] == request_data["inputs"] + + # Verify session_id is present when provided in inputs + if "session_id" in result: + assert result["session_id"] == "session-456" + + finally: + async with session_scope() as session: + flow = await session.get(Flow, flow_id) + if flow: + await session.delete(flow) + + async def test_sync_execution_with_llm_output( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """Test sync execution with LLM component output including model metadata.""" + flow_id = uuid4() + + async with session_scope() as session: + flow = Flow( + id=flow_id, + name="LLM Flow", + description="Flow with LLM component", + data={"nodes": [], "edges": []}, + user_id=created_api_key.user_id, + ) + session.add(flow) + await session.flush() + await session.refresh(flow) + + try: + request_data = { + "flow_id": str(flow_id), + "background": False, + "stream": False, + "inputs": { + "ChatInput-abc.input_value": "Explain quantum computing", + "OpenAIModel-def.temperature": 0.7, + }, + } + + # Mock LLM execution with model metadata + mock_result_data = MagicMock() + mock_result_data.component_id = "OpenAIModel-def" + mock_result_data.outputs = { + "model_output": { + "message": { + "message": "Quantum computing uses quantum mechanics...", + "model_name": "gpt-4", + "type": "text", + } + } + } + mock_result_data.metadata = {"tokens_used": 150} + + # Wrap ResultData in RunOutputs + mock_run_output = MagicMock() + mock_run_output.outputs = [mock_result_data] + + with patch("langflow.api.v2.workflow.run_graph_internal") as mock_run: + # run_graph_internal returns tuple[list[RunOutputs], str] + mock_run.return_value = ([mock_run_output], "session-789") + + headers = {"x-api-key": created_api_key.api_key} + response = await client.post("api/v2/workflow", json=request_data, headers=headers) + + assert response.status_code == 200 + result = response.json() + + assert result["flow_id"] == str(flow_id) + assert "job_id" in result + assert "outputs" in result + assert "metadata" in result + # Note: Detailed content validation requires proper graph/vertex mocking + + finally: + async with session_scope() as session: + flow = await session.get(Flow, flow_id) + if flow: + await session.delete(flow) + + async def test_sync_execution_with_file_save_output( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """Test sync execution with SaveToFile component.""" + flow_id = uuid4() + + async with session_scope() as session: + flow = Flow( + id=flow_id, + name="File Save Flow", + description="Flow with file save component", + data={"nodes": [], "edges": []}, + user_id=created_api_key.user_id, + ) + session.add(flow) + await session.flush() + await session.refresh(flow) + + try: + request_data = { + "flow_id": str(flow_id), + "background": False, + "stream": False, + "inputs": { + "TextInput-abc.text": "Content to save", + "SaveToFile-xyz.file_path": "/tmp/output.txt", # noqa: S108 + }, + } + + # Mock SaveToFile execution + mock_result_data = MagicMock() + mock_result_data.component_id = "SaveToFile-xyz" + mock_result_data.outputs = { + "message": {"message": "File saved successfully to /tmp/output.txt", "type": "text"} + } + mock_result_data.metadata = {"bytes_written": 1024} + + # Wrap ResultData in RunOutputs + mock_run_output = MagicMock() + mock_run_output.outputs = [mock_result_data] + + with patch("langflow.api.v2.workflow.run_graph_internal") as mock_run: + # run_graph_internal returns tuple[list[RunOutputs], str] + mock_run.return_value = ([mock_run_output], "session-101") + + headers = {"x-api-key": created_api_key.api_key} + response = await client.post("api/v2/workflow", json=request_data, headers=headers) + + assert response.status_code == 200 + result = response.json() + + assert result["flow_id"] == str(flow_id) + assert "job_id" in result + assert "outputs" in result + assert "metadata" in result + # Note: Detailed content validation requires proper graph/vertex mocking + + finally: + async with session_scope() as session: + flow = await session.get(Flow, flow_id) + if flow: + await session.delete(flow) + + async def test_sync_execution_with_multiple_terminal_nodes( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """Test sync execution with multiple terminal nodes (outputs).""" + flow_id = uuid4() + + async with session_scope() as session: + flow = Flow( + id=flow_id, + name="Multi-Output Flow", + description="Flow with multiple terminal nodes", + data={"nodes": [], "edges": []}, + user_id=created_api_key.user_id, + ) + session.add(flow) + await session.flush() + await session.refresh(flow) + + try: + request_data = { + "flow_id": str(flow_id), + "background": False, + "stream": False, + "inputs": {"ChatInput-abc.input_value": "Process this"}, + } + + # Mock execution with multiple outputs + mock_chat_output = MagicMock() + mock_chat_output.component_id = "ChatOutput-aaa" + mock_chat_output.outputs = {"message": {"message": "Chat response", "type": "text"}} + + mock_file_output = MagicMock() + mock_file_output.component_id = "SaveToFile-bbb" + mock_file_output.outputs = {"message": {"message": "File saved successfully", "type": "text"}} + + with patch("langflow.api.v2.workflow.run_graph_internal") as mock_run: + # run_graph_internal returns tuple[list[RunOutputs], str] + mock_run.return_value = ([mock_chat_output, mock_file_output], "session-202") + + headers = {"x-api-key": created_api_key.api_key} + response = await client.post("api/v2/workflow", json=request_data, headers=headers) + + assert response.status_code == 200 + result = response.json() + + assert result["flow_id"] == str(flow_id) + assert "job_id" in result + assert "outputs" in result + # Note: Detailed content validation requires proper graph/vertex mocking + + finally: + async with session_scope() as session: + flow = await session.get(Flow, flow_id) + if flow: + await session.delete(flow) + + async def test_sync_execution_response_structure_validation( + self, + client: AsyncClient, + created_api_key, + mock_settings_dev_api_enabled, # noqa: ARG002 + ): + """Test that sync execution response has correct WorkflowExecutionResponse structure.""" + flow_id = uuid4() + + async with session_scope() as session: + flow = Flow( + id=flow_id, + name="Test Flow", + description="Flow for response validation", + data={"nodes": [], "edges": []}, + user_id=created_api_key.user_id, + ) + session.add(flow) + await session.flush() + await session.refresh(flow) + + try: + request_data = {"flow_id": str(flow_id), "background": False, "stream": False, "inputs": None} + + headers = {"x-api-key": created_api_key.api_key} + response = await client.post("api/v2/workflow", json=request_data, headers=headers) + + assert response.status_code == 200 + result = response.json() + + # Verify WorkflowExecutionResponse structure + assert "flow_id" in result + assert isinstance(result["flow_id"], str) + assert result["flow_id"] == str(flow_id) + + assert "job_id" in result + assert isinstance(result["job_id"], str) + + # session_id is optional - only present if provided in inputs + if "session_id" in result: + assert isinstance(result["session_id"], str) + + assert "object" in result + assert result["object"] == "response" + + assert "created_timestamp" in result + assert isinstance(result["created_timestamp"], str) + + assert "status" in result + assert result["status"] in ["completed", "failed", "running", "queued"] + + assert "errors" in result + assert isinstance(result["errors"], list) + + assert "inputs" in result + assert isinstance(result["inputs"], dict) + + assert "outputs" in result + assert isinstance(result["outputs"], dict) + + assert "metadata" in result + assert isinstance(result["metadata"], dict) + + finally: + async with session_scope() as session: + flow = await session.get(Flow, flow_id) + if flow: + await session.delete(flow) + # Test POST /workflow/stop without API key response = await client.post( "api/v2/workflow/stop", diff --git a/src/lfx/src/lfx/schema/workflow.py b/src/lfx/src/lfx/schema/workflow.py index 0c766055fd..c8ce1c408e 100644 --- a/src/lfx/src/lfx/schema/workflow.py +++ b/src/lfx/src/lfx/schema/workflow.py @@ -36,14 +36,6 @@ class ComponentOutput(BaseModel): metadata: dict[str, Any] | None = None -class GlobalInputs(BaseModel): - """Global inputs that apply to all input components in the workflow.""" - - input_value: str | None = Field(None, description="The input value to send to input components") - input_type: str = Field("chat", description="The type of input (chat, text, etc.)") - session_id: str | None = Field(None, description="Session ID for conversation continuity") - - class WorkflowExecutionRequest(BaseModel): """Request schema for workflow execution.""" @@ -51,7 +43,7 @@ class WorkflowExecutionRequest(BaseModel): stream: bool = False flow_id: str inputs: dict[str, Any] | None = Field( - None, description="Inputs with 'global' key for global inputs and component IDs for component-specific tweaks" + None, description="Component-specific inputs in flat format: 'component_id.param_name': value" ) model_config = ConfigDict( @@ -62,26 +54,28 @@ class WorkflowExecutionRequest(BaseModel): "stream": False, "flow_id": "flow_67ccd2be17f0819081ff3bb2cf6508e60bb6a6b452d3795b", "inputs": { - "global": { - "input_value": "Hello, how can you help me today?", - "input_type": "chat", - "session_id": "session-123", - }, - "llm_component": {"temperature": 0.7, "max_tokens": 100}, - "opensearch_component": {"opensearch_url": "https://opensearch:9200"}, + "ChatInput-abc.input_value": "Hello, how can you help me today?", + "ChatInput-abc.session_id": "session-123", + "LLM-xyz.temperature": 0.7, + "LLM-xyz.max_tokens": 100, + "OpenSearch-def.opensearch_url": "https://opensearch:9200", }, }, { "background": True, "stream": False, "flow_id": "flow_67ccd2be17f0819081ff3bb2cf6508e60bb6a6b452d3795b", - "inputs": {"global": {"input_value": "Process this in the background", "input_type": "text"}}, + "inputs": { + "ChatInput-abc.input_value": "Process this in the background", + }, }, { "background": False, "stream": True, "flow_id": "flow_67ccd2be17f0819081ff3bb2cf6508e60bb6a6b452d3795b", - "inputs": {"chat_component": {"text": "Stream this conversation"}}, + "inputs": { + "ChatInput-abc.input_value": "Stream this conversation", + }, }, ] }, @@ -94,7 +88,7 @@ class WorkflowExecutionResponse(BaseModel): flow_id: str job_id: str - object: Literal["response"] = "response" + object: Literal["response"] = Field(default="response") created_timestamp: str status: JobStatus errors: list[ErrorDetail] = [] @@ -107,6 +101,7 @@ class WorkflowJobResponse(BaseModel): """Background job response.""" job_id: str + object: Literal["job"] = Field(default="job") created_timestamp: str status: JobStatus errors: list[ErrorDetail] = [] @@ -146,7 +141,14 @@ WORKFLOW_EXECUTION_RESPONSES = { "oneOf": [ WorkflowExecutionResponse.model_json_schema(), WorkflowJobResponse.model_json_schema(), - ] + ], + "discriminator": { + "propertyName": "object", + "mapping": { + "response": "#/components/schemas/WorkflowExecutionResponse", + "job": "#/components/schemas/WorkflowJobResponse", + }, + }, } }, "text/event-stream": {