feat(api/v2): add request-side output selection (output_ids)

Let a sync caller name the output(s) they want via output_ids so
output.text resolves deterministically (reason=single) on multi-output
flows instead of going null. Selection is steer-only: it picks the
answer among the named outputs without filtering the outputs map.

Invalid ids are rejected with 422 before the flow runs (and before any
job row is created), so a typo costs no compute. Resolution considers
selected outputs that actually fired, so branching flows resolve to
whichever candidate ran.

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,
     )
 

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:

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])

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"]

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."""
 

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=(

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)