diff --git a/src/backend/base/langflow/api/v2/converters.py b/src/backend/base/langflow/api/v2/converters.py index dbf06fc167..9fbcb63a1e 100644 --- a/src/backend/base/langflow/api/v2/converters.py +++ b/src/backend/base/langflow/api/v2/converters.py @@ -56,6 +56,10 @@ class ParsedWorkflowRun: mode: str = "stream" start_component_id: str | None = None stop_component_id: str | None = None + # Request-side output selection (sync mode): when set, the answer (``output``) + # resolves among only these component ids. Validated against the flow's outputs + # before the flow runs. + output_ids: list[str] | None = None # Optional live flow data (nodes + edges) overriding the DB copy. Lets the # canvas run with unsaved tweaks the user has made but not yet persisted. data: dict[str, Any] | None = None @@ -87,6 +91,7 @@ def parse_workflow_run_request(request: WorkflowRunRequest) -> ParsedWorkflowRun mode=request.mode.value, start_component_id=request.start_component_id, stop_component_id=request.stop_component_id, + output_ids=request.output_ids, data=request.data, files=request.files, globals=dict(request.globals or {}), @@ -383,7 +388,7 @@ def _process_terminal_vertex( return output_key, component_output -def _resolve_output(outputs: dict[str, ComponentOutput]) -> WorkflowOutput: +def _resolve_output(outputs: dict[str, ComponentOutput], selected_ids: list[str] | None = None) -> WorkflowOutput: """Resolve the run's primary text answer and explain why it resolved that way. ``text`` is the lone string answer only when exactly one output is a text @@ -396,12 +401,24 @@ def _resolve_output(outputs: dict[str, ComponentOutput]) -> WorkflowOutput: - ``non_string``: a text channel exists but its content isn't a string - ``none``: no text channel at all (e.g. a data-only flow) + ``selected_ids`` is the request-side ``output_ids`` selection. When given (and + non-empty), resolution considers only the named outputs that this run actually + produced (selected ∩ produced), so naming one ChatOutput on a multi-output flow + yields ``single`` deterministically. Names that didn't fire this run (e.g. a + branch not taken) are simply absent. ``outputs`` itself is never filtered — the + selection steers ``output.text``, it does not hide outputs. An empty list means + "no selection", same as None. + An intentionally empty single answer is preserved as "" (distinct from None). The ``failed`` reason is set on the error path, not here. """ + considered = outputs + if selected_ids: + selected = set(selected_ids) + considered = {component_id: output for component_id, output in outputs.items() if component_id in selected} text_items = [ (component_id, output.content) - for component_id, output in outputs.items() + for component_id, output in considered.items() if output.type in {"message", "text"} and isinstance(output.content, str) ] if len(text_items) == 1: @@ -412,7 +429,7 @@ def _resolve_output(outputs: dict[str, ComponentOutput]) -> WorkflowOutput: return WorkflowOutput(reason=OutputReason.MULTIPLE) # Zero string text answers: distinguish "a text channel exists but its # content isn't a string" from "no text channel at all". - has_text_channel = any(output.type in {"message", "text"} for output in outputs.values()) + has_text_channel = any(output.type in {"message", "text"} for output in considered.values()) reason = OutputReason.NON_STRING if has_text_channel else OutputReason.NONE return WorkflowOutput(reason=reason) @@ -424,6 +441,7 @@ def run_response_to_workflow_response( inputs: dict[str, Any], graph: Graph, effective_globals: dict[str, str] | None = None, + selected_ids: list[str] | None = None, ) -> WorkflowExecutionResponse: """Convert V1 RunResponse to V2 WorkflowExecutionResponse. @@ -438,9 +456,10 @@ def run_response_to_workflow_response( - 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 + - Keys ``outputs`` by the stable ``vertex.id`` (component id), never the + display name (a display-name alias would collide and break on rename). + - ``display_name`` is carried as a field on each ``ComponentOutput`` for + human-readable rendering. Args: run_response: The V1 response from simple_run_flow containing execution results @@ -451,6 +470,8 @@ def run_response_to_workflow_response( effective_globals: The ``globals`` echoed in the response — the effective merged set (body globals plus any legacy header-supplied globals). When omitted, an empty globals map is echoed. + selected_ids: The request-side ``output_ids`` selection. When set, ``output`` + resolves among only these component ids; ``outputs`` is unaffected. Returns: WorkflowExecutionResponse: V2 schema response with structured outputs @@ -502,7 +523,7 @@ def run_response_to_workflow_response( errors=[], inputs=inputs or {}, globals=response_globals, - output=_resolve_output(outputs), + output=_resolve_output(outputs, selected_ids=selected_ids), outputs=outputs, ) diff --git a/src/backend/base/langflow/api/v2/workflow.py b/src/backend/base/langflow/api/v2/workflow.py index 34714201d5..f156635c38 100644 --- a/src/backend/base/langflow/api/v2/workflow.py +++ b/src/backend/base/langflow/api/v2/workflow.py @@ -113,6 +113,30 @@ def _unknown_protocol_http_exception(exc: UnknownStreamProtocolError) -> HTTPExc ) +def _validate_output_ids(output_ids: list[str] | None, terminal_node_ids: list[str]) -> None: + """Reject ``output_ids`` that aren't outputs of this flow, BEFORE it runs. + + Checks against the terminal node ids known after graph build but before + execution, so a typo or wrong id costs no compute. ``available`` lists the + flow's real output ids so callers can self-correct. None/empty means "no + selection" and never raises. + """ + if not output_ids: + return + known = set(terminal_node_ids) + unknown = [output_id for output_id in output_ids if output_id not in known] + if unknown: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail={ + "error": "Unknown output_ids", + "code": "UNKNOWN_OUTPUT_IDS", + "message": f"output_ids not produced by this flow: {unknown}.", + "available": terminal_node_ids, + }, + ) + + def _build_run_inputs(parsed: ParsedWorkflowRun) -> list[InputValueRequest] | None: """Build the graph input list from the AG-UI chat message, if any. @@ -446,6 +470,11 @@ async def execute_sync_workflow( # Get terminal nodes - these are the outputs we want terminal_node_ids = graph.get_terminal_nodes() + # Validate request-side output selection BEFORE executing: a bad id must cost + # no compute. Raised outside the component-error try/except below, so it + # surfaces as a real 422 rather than a 200-with-failed body. + _validate_output_ids(parsed.output_ids, terminal_node_ids) + # Execute graph - component errors are caught and returned in response body job_service = get_job_service() await job_service.create_job(job_id=job_id, flow_id=flow_id_str, user_id=current_user.id) @@ -483,6 +512,7 @@ async def execute_sync_workflow( inputs=parsed.tweaks, graph=graph, effective_globals=request_variables, + selected_ids=parsed.output_ids, ) except asyncio.CancelledError: diff --git a/src/backend/tests/unit/api/v2/test_converters.py b/src/backend/tests/unit/api/v2/test_converters.py index d3f1a85b29..38dd24be3f 100644 --- a/src/backend/tests/unit/api/v2/test_converters.py +++ b/src/backend/tests/unit/api/v2/test_converters.py @@ -1248,6 +1248,64 @@ class TestResolveOutput: assert output.reason == OutputReason.NONE assert output.text is None + # --- request-side output selection (steer-only) --- + + def test_selected_id_disambiguates_two_messages_to_single(self): + # Two text outputs would be ``multiple``; naming one steers ``output.text`` + # to it deterministically without hiding the other from ``outputs``. + outputs = { + "ChatOutput-a": _component_output("message", "Hi"), + "ChatOutput-b": _component_output("message", "Bye"), + } + output = _resolve_output(outputs, selected_ids=["ChatOutput-b"]) + assert output.reason == OutputReason.SINGLE + assert output.text == "Bye" + assert output.source == "ChatOutput-b" + + def test_selecting_both_text_outputs_stays_multiple(self): + outputs = { + "ChatOutput-a": _component_output("message", "Hi"), + "ChatOutput-b": _component_output("message", "Bye"), + } + output = _resolve_output(outputs, selected_ids=["ChatOutput-a", "ChatOutput-b"]) + assert output.reason == OutputReason.MULTIPLE + assert output.text is None + + def test_selected_id_not_produced_resolves_among_produced(self): + # Branching: the caller lists candidate outputs; the one that didn't fire is + # simply absent from ``outputs``, so we resolve among the ones that did. + outputs = { + "ChatOutput-a": _component_output("message", "the reply"), + "DataOutput-c": _component_output("data", {"k": "v"}), + } + output = _resolve_output(outputs, selected_ids=["ChatOutput-a", "ChatOutput-b"]) + assert output.reason == OutputReason.SINGLE + assert output.text == "the reply" + assert output.source == "ChatOutput-a" + + def test_selection_with_empty_intersection_is_none(self): + # Every named id is absent from this run's outputs -> nothing to resolve. + outputs = {"ChatOutput-a": _component_output("message", "Hi")} + output = _resolve_output(outputs, selected_ids=["ChatOutput-z"]) + assert output.reason == OutputReason.NONE + assert output.text is None + + def test_no_selection_preserves_default_resolution(self): + # selected_ids=None keeps the original behavior (two text outputs -> multiple). + outputs = { + "ChatOutput-a": _component_output("message", "Hi"), + "ChatOutput-b": _component_output("message", "Bye"), + } + assert _resolve_output(outputs, selected_ids=None).reason == OutputReason.MULTIPLE + + def test_empty_selection_list_is_treated_as_no_selection(self): + # An empty list is not "select nothing"; it means no filter (same as None). + outputs = { + "ChatOutput-a": _component_output("message", "Hi"), + "ChatOutput-b": _component_output("message", "Bye"), + } + assert _resolve_output(outputs, selected_ids=[]).reason == OutputReason.MULTIPLE + def _message_output_vertex(vertex_id: str) -> Mock: vertex = Mock() @@ -1363,6 +1421,32 @@ class TestOutputAndSessionId: assert response.outputs["ChatOutput-a"].content == "Hi" assert response.outputs["ChatOutput-b"].content == "Bye" + def test_output_ids_selection_steers_two_outputs_to_single(self): + # Two ChatOutputs would resolve to ``multiple``; selecting one steers + # ``output.text`` to it while ``outputs`` still carries both. + vertices = [_message_output_vertex("ChatOutput-a"), _message_output_vertex("ChatOutput-b")] + graph = _graph_for(vertices) + + run_response = Mock() + run_response.session_id = "session-xyz" + run_output = Mock() + run_output.outputs = [ + _message_result_data("ChatOutput-a", "Hi"), + _message_result_data("ChatOutput-b", "Bye"), + ] + run_response.outputs = [run_output] + + response = run_response_to_workflow_response( + run_response, "flow-1", str(uuid4()), {}, graph, selected_ids=["ChatOutput-b"] + ) + + assert response.output.reason == OutputReason.SINGLE + assert response.output.text == "Bye" + assert response.output.source == "ChatOutput-b" + # Selection steers the answer but never hides the other output. + assert response.outputs["ChatOutput-a"].content == "Hi" + assert response.outputs["ChatOutput-b"].content == "Bye" + def test_session_id_none_passes_through(self): vertex = _message_output_vertex("ChatOutput-abc") graph = _graph_for([vertex]) diff --git a/src/backend/tests/unit/api/v2/test_workflow.py b/src/backend/tests/unit/api/v2/test_workflow.py index bc54b1f476..6672e7201d 100644 --- a/src/backend/tests/unit/api/v2/test_workflow.py +++ b/src/backend/tests/unit/api/v2/test_workflow.py @@ -740,3 +740,36 @@ class TestWorkflowIDORProtection: db_flow = await session.get(Flow, flow_id) if db_flow: await session.delete(db_flow) + + +class TestValidateOutputIds: + """``output_ids`` is validated against the flow's outputs BEFORE the flow runs. + + The validator only ever sees the terminal node ids (known after graph build, + before execution), so a bad id is rejected with 422 without spending any + compute. This is the pure-function contract; the route places the call before + execution. + """ + + def test_no_selection_is_a_noop(self): + from langflow.api.v2.workflow import _validate_output_ids + + # None and empty both mean "no selection" -> never raises. + _validate_output_ids(None, ["ChatOutput-a"]) + _validate_output_ids([], ["ChatOutput-a"]) + + def test_all_known_ids_pass(self): + from langflow.api.v2.workflow import _validate_output_ids + + _validate_output_ids(["ChatOutput-a"], ["ChatOutput-a", "ChatOutput-b"]) + + def test_unknown_id_raises_422_listing_available(self): + from fastapi import HTTPException + from langflow.api.v2.workflow import _validate_output_ids + + with pytest.raises(HTTPException) as exc: + _validate_output_ids(["ChatOutput-z"], ["ChatOutput-a", "ChatOutput-b"]) + assert exc.value.status_code == 422 + detail = exc.value.detail + assert "ChatOutput-z" in str(detail) + assert detail["available"] == ["ChatOutput-a", "ChatOutput-b"] diff --git a/src/backend/tests/unit/api/v2/test_workflow_agui.py b/src/backend/tests/unit/api/v2/test_workflow_agui.py index 3a9a2abb23..a8b43cc193 100644 --- a/src/backend/tests/unit/api/v2/test_workflow_agui.py +++ b/src/backend/tests/unit/api/v2/test_workflow_agui.py @@ -331,6 +331,63 @@ class TestAGUISyncExecution: assert result["outputs"] +class TestOutputIdsSelection: + """Request-side ``output_ids`` steers ``output.text`` and is validated pre-run.""" + + async def test_unknown_output_id_rejected_before_running( + self, + client: AsyncClient, + created_api_key, + chatbot_flow, + ): + """A typo'd output id is rejected with 422 and the available ids, no run.""" + body = _agui_body(chatbot_flow, message="hi", mode="sync") + body["output_ids"] = ["NotARealOutput-zzz"] + + response = await client.post( + "api/v2/workflows", + json=body, + headers={"x-api-key": created_api_key.api_key}, + ) + + # 422 (a request rejection), NOT a 200-with-failed component error. + assert response.status_code == 422 + detail = response.json()["detail"] + assert "NotARealOutput-zzz" in str(detail) + assert detail["available"] # the flow's real output ids + + async def test_selecting_the_real_output_yields_single( + self, + client: AsyncClient, + created_api_key, + chatbot_flow, + ): + """Naming the flow's text output makes ``output.reason`` deterministically single.""" + headers = {"x-api-key": created_api_key.api_key} + + # Discover the flow's text output id from an unselected run. + first = await client.post( + "api/v2/workflows", + json=_agui_body(chatbot_flow, message="hi", mode="sync"), + headers=headers, + ) + assert first.status_code == 200 + outputs = first.json()["outputs"] + text_id = next(cid for cid, out in outputs.items() if out["type"] in {"message", "text"}) + + body = _agui_body(chatbot_flow, message="hi again", mode="sync") + body["output_ids"] = [text_id] + second = await client.post("api/v2/workflows", json=body, headers=headers) + + assert second.status_code == 200 + result = second.json() + assert result["output"]["reason"] == "single" + assert result["output"]["source"] == text_id + assert result["output"]["text"] is not None + # Selection steers the answer; the full outputs map is still present. + assert text_id in result["outputs"] + + class TestAGUICancellation: """A run can be stopped by job id, and a streaming client can disconnect.""" diff --git a/src/lfx/src/lfx/schema/workflow.py b/src/lfx/src/lfx/schema/workflow.py index d7cbbbcb6d..cd37bddf3a 100644 --- a/src/lfx/src/lfx/schema/workflow.py +++ b/src/lfx/src/lfx/schema/workflow.py @@ -219,6 +219,16 @@ class WorkflowRunRequest(BaseModel): ) start_component_id: str | None = Field(None, description="Partial-run start component id.") stop_component_id: str | None = Field(None, description="Partial-run stop component id.") + output_ids: list[str] | None = Field( + None, + description=( + "Component ids of the outputs you want as the answer (sync mode). When set, " + "``output.text`` resolves among only these, so naming one text output makes " + "``output.reason`` deterministic on multi-output flows. The full ``outputs`` " + "map is still returned. Ids must be outputs of this flow or the request is " + "rejected before the flow runs. Ignored for stream/background modes." + ), + ) globals: dict[GlobalVarKey, GlobalVarValue] = Field( default_factory=dict, description=( diff --git a/src/lfx/tests/unit/schema/test_workflow_run_request.py b/src/lfx/tests/unit/schema/test_workflow_run_request.py index ddb305d729..4a632dddfc 100644 --- a/src/lfx/tests/unit/schema/test_workflow_run_request.py +++ b/src/lfx/tests/unit/schema/test_workflow_run_request.py @@ -24,6 +24,11 @@ class TestDefaults: assert req.files is None assert req.start_component_id is None assert req.stop_component_id is None + assert req.output_ids is None + + def test_output_ids_accepts_component_id_list(self): + req = WorkflowRunRequest(flow_id=_VALID_UUID, output_ids=["ChatOutput-abc", "ChatOutput-def"]) + assert req.output_ids == ["ChatOutput-abc", "ChatOutput-def"] def test_mode_accepts_string_value(self): req = WorkflowRunRequest(flow_id=_VALID_UUID, mode="stream") @@ -90,6 +95,7 @@ class TestRoundTripsWithRichBody: "files": ["/tmp/a.txt", "/tmp/b.png"], "start_component_id": "ChatInput-abc", "stop_component_id": "ChatOutput-xyz", + "output_ids": ["ChatOutput-xyz"], "globals": {"API_TOKEN": "secret-123"}, } req = WorkflowRunRequest.model_validate(body)