feat(loop): implement isolated subgraph execution for LoopComponent (#11034)

* refactor(loop): implement isolated subgraph execution for LoopComponent

- Refactor LoopComponent to execute loop body as isolated subgraph
- Add create_subgraph method to Graph class for creating isolated subgraphs
- Add loop_utils with get_loop_body_vertices helper function
- Add on_end_vertex event emission in async_start for progress tracking
- Add comprehensive tests for subgraph execution and event emission

This change enables proper vertex event streaming during loop execution
by running each iteration as an isolated subgraph with its own context.

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

* [autofix.ci] apply automated fixes (attempt 3/3)

* fix: Improve subgraph event streaming to UI

- Add event types to stream token event manager (end_vertex, error, build_start, build_end)
- Preserve start_component_id in async_start for proper subgraph execution
- Fix has_chat_output/has_chat_input to use vertex.base_name instead of string matching
- Add _stream_to_playground check to allow inner graph components to send events
- Fix _send_message_event to properly extract and forward message IDs

* update index

* fix: Enhance extract_loop_output to handle multiple message formats

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

* [autofix.ci] apply automated fixes (attempt 3/3)

* fix: Improve intra-layer cycle detection logging in sorting function

* refactor: Update loop execution to create fresh subgraph for each iteration

* test: Add tests for subgraph isolation and state management

* fix: Update LoopComponent description for clarity on item processing

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

* [autofix.ci] apply automated fixes (attempt 3/3)

* fix: Add warning log for missing custom_component in loop item injection

* refactor: Simplify loop event tests while preserving critical coverage

Reduced test complexity by removing excessive mocking while maintaining completeå event manager propagation coverage. Event manager tests are critical for ensuring UI updates work correctly during loop execution.

* fix: Update Google dependency version to 2.5.0 across multiple components

* fix: Update value format in Research Translation Loop JSON for clarity

* fix: Inject loop items into HandleInput fields via raw_params

HandleInput fields (type="other") were receiving None instead of loop items
because:
1. Fields with type="other" are skipped during field param processing
2. Loop->Parser edge is filtered out in subgraph creation
3. updated_raw_params flag was reset too early by multiple build_params() calls

Fix:
- Inject loop items into template before prepare()
- Inject into raw_params after prepare() using update_raw_params()
- Persist updated_raw_params flag across all build_params() calls
- Reset flag in _build_each_vertex_in_params_dict() after processing

Tests:
- Add integration test exercising execute_loop_body() code path
- Verify update_raw_params() called with correct data per iteration
- Add full Loop+Parser integration test
- Fix fieldName -> field_name in existing tests

* fix: Update done_output method to use internal event manager and improve event handling

* refactor: Update execute_loop_body to use async context manager for subgraph creation and cleanup

* fix: import json in unified_models.py

* feat: Implement subgraph tracing context management

* fix: Update code_hash and improve loop component execution logic

* fix: Update code_hash and refine LoopComponent execution logic

* [autofix.ci] apply automated fixes

* fix: Update LoopComponent subgraph execution to set event manager as instance attribute

* test: Enhance subgraph execution tests to verify event manager passing

---------

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

Author: ogabrielluiz

src/backend/base/langflow/initial_setup/starter_projects/Research Translation Loop.json

@@ -1449,7 +1449,7 @@
             "beta": false,
             "conditional_paths": [],
             "custom_fields": {},
-            "description": "Iterates over a list of Data or Message objects, outputting one item at a time and aggregating results from loop inputs. Message objects are automatically converted to Data objects for consistent processing.",
+            "description": "Iterates over a list of Data or Message objects, processing one item at a time and aggregating results from loop inputs. Message objects are automatically converted to Data objects for consistent processing.",
             "display_name": "Loop",
             "documentation": "https://docs.langflow.org/loop",
             "edited": false,
@@ -1460,7 +1460,7 @@
             "icon": "infinity",
             "legacy": false,
             "metadata": {
-              "code_hash": "e179036a232d",
+              "code_hash": "9f88705635cd",
               "dependencies": {
                 "dependencies": [
                   {
@@ -1526,7 +1526,7 @@
                 "show": true,
                 "title_case": false,
                 "type": "code",
-                "value": "from lfx.components.processing.converter import convert_to_data\nfrom lfx.custom.custom_component.component import Component\nfrom lfx.inputs.inputs import HandleInput\nfrom lfx.schema.data import Data\nfrom lfx.schema.dataframe import DataFrame\nfrom lfx.schema.message import Message\nfrom lfx.template.field.base import Output\n\n\nclass LoopComponent(Component):\n    display_name = \"Loop\"\n    description = (\n        \"Iterates over a list of Data or Message objects, outputting one item at a time and \"\n        \"aggregating results from loop inputs. Message objects are automatically converted to \"\n        \"Data objects for consistent processing.\"\n    )\n    documentation: str = \"https://docs.langflow.org/loop\"\n    icon = \"infinity\"\n\n    inputs = [\n        HandleInput(\n            name=\"data\",\n            display_name=\"Inputs\",\n            info=\"The initial DataFrame to iterate over.\",\n            input_types=[\"DataFrame\"],\n        ),\n    ]\n\n    outputs = [\n        Output(\n            display_name=\"Item\",\n            name=\"item\",\n            method=\"item_output\",\n            allows_loop=True,\n            loop_types=[\"Message\"],\n            group_outputs=True,\n        ),\n        Output(display_name=\"Done\", name=\"done\", method=\"done_output\", group_outputs=True),\n    ]\n\n    def initialize_data(self) -> None:\n        \"\"\"Initialize the data list, context index, and aggregated list.\"\"\"\n        if self.ctx.get(f\"{self._id}_initialized\", False):\n            return\n\n        # Ensure data is a list of Data objects\n        data_list = self._validate_data(self.data)\n\n        # Store the initial data and context variables\n        self.update_ctx(\n            {\n                f\"{self._id}_data\": data_list,\n                f\"{self._id}_index\": 0,\n                f\"{self._id}_aggregated\": [],\n                f\"{self._id}_initialized\": True,\n            }\n        )\n\n    def _convert_message_to_data(self, message: Message) -> Data:\n        \"\"\"Convert a Message object to a Data object using Type Convert logic.\"\"\"\n        return convert_to_data(message, auto_parse=False)\n\n    def _validate_data(self, data):\n        \"\"\"Validate and return a list of Data objects. Message objects are auto-converted to Data.\"\"\"\n        if isinstance(data, DataFrame):\n            return data.to_data_list()\n        if isinstance(data, Data):\n            return [data]\n        if isinstance(data, Message):\n            # Auto-convert Message to Data\n            converted_data = self._convert_message_to_data(data)\n            return [converted_data]\n        if isinstance(data, list) and all(isinstance(item, (Data, Message)) for item in data):\n            # Convert any Message objects in the list to Data objects\n            converted_list = []\n            for item in data:\n                if isinstance(item, Message):\n                    converted_list.append(self._convert_message_to_data(item))\n                else:\n                    converted_list.append(item)\n            return converted_list\n        msg = \"The 'data' input must be a DataFrame, a list of Data/Message objects, or a single Data/Message object.\"\n        raise TypeError(msg)\n\n    def evaluate_stop_loop(self) -> bool:\n        \"\"\"Evaluate whether to stop item or done output.\"\"\"\n        current_index = self.ctx.get(f\"{self._id}_index\", 0)\n        data_length = len(self.ctx.get(f\"{self._id}_data\", []))\n        return current_index > data_length\n\n    def item_output(self) -> Data:\n        \"\"\"Output the next item in the list or stop if done.\"\"\"\n        self.initialize_data()\n        current_item = Data(text=\"\")\n\n        if self.evaluate_stop_loop():\n            self.stop(\"item\")\n        else:\n            # Get data list and current index\n            data_list, current_index = self.loop_variables()\n            if current_index < len(data_list):\n                # Output current item and increment index\n                try:\n                    current_item = data_list[current_index]\n                except IndexError:\n                    current_item = Data(text=\"\")\n            self.aggregated_output()\n            self.update_ctx({f\"{self._id}_index\": current_index + 1})\n\n        # Now we need to update the dependencies for the next run\n        self.update_dependency()\n        return current_item\n\n    def update_dependency(self):\n        item_dependency_id = self.get_incoming_edge_by_target_param(\"item\")\n        if item_dependency_id not in self.graph.run_manager.run_predecessors[self._id]:\n            self.graph.run_manager.run_predecessors[self._id].append(item_dependency_id)\n            # CRITICAL: Also update run_map so remove_from_predecessors() works correctly\n            # run_map[predecessor] = list of vertices that depend on predecessor\n            if self._id not in self.graph.run_manager.run_map[item_dependency_id]:\n                self.graph.run_manager.run_map[item_dependency_id].append(self._id)\n\n    def done_output(self) -> DataFrame:\n        \"\"\"Trigger the done output when iteration is complete.\"\"\"\n        self.initialize_data()\n\n        if self.evaluate_stop_loop():\n            self.stop(\"item\")\n            self.start(\"done\")\n\n            aggregated = self.ctx.get(f\"{self._id}_aggregated\", [])\n\n            return DataFrame(aggregated)\n        self.stop(\"done\")\n        return DataFrame([])\n\n    def loop_variables(self):\n        \"\"\"Retrieve loop variables from context.\"\"\"\n        return (\n            self.ctx.get(f\"{self._id}_data\", []),\n            self.ctx.get(f\"{self._id}_index\", 0),\n        )\n\n    def aggregated_output(self) -> list[Data]:\n        \"\"\"Return the aggregated list once all items are processed.\n\n        Returns Data or Message objects depending on loop input types.\n        \"\"\"\n        self.initialize_data()\n\n        # Get data list and aggregated list\n        data_list = self.ctx.get(f\"{self._id}_data\", [])\n        aggregated = self.ctx.get(f\"{self._id}_aggregated\", [])\n        loop_input = self.item\n\n        # Append the current loop input to aggregated if it's not already included\n        if loop_input is not None and not isinstance(loop_input, str) and len(aggregated) <= len(data_list):\n            # If the loop input is a Message, convert it to Data for consistency\n            if isinstance(loop_input, Message):\n                loop_input = self._convert_message_to_data(loop_input)\n            aggregated.append(loop_input)\n            self.update_ctx({f\"{self._id}_aggregated\": aggregated})\n        return aggregated\n"
+                "value": "from lfx.base.flow_controls.loop_utils import (\n    execute_loop_body,\n    extract_loop_output,\n    get_loop_body_start_edge,\n    get_loop_body_start_vertex,\n    get_loop_body_vertices,\n    validate_data_input,\n)\nfrom lfx.components.processing.converter import convert_to_data\nfrom lfx.custom.custom_component.component import Component\nfrom lfx.inputs.inputs import HandleInput\nfrom lfx.schema.data import Data\nfrom lfx.schema.dataframe import DataFrame\nfrom lfx.schema.message import Message\nfrom lfx.template.field.base import Output\n\n\nclass LoopComponent(Component):\n    display_name = \"Loop\"\n    description = (\n        \"Iterates over a list of Data or Message objects, processing one item at a time and \"\n        \"aggregating results from loop inputs. Message objects are automatically converted to \"\n        \"Data objects for consistent processing.\"\n    )\n    documentation: str = \"https://docs.langflow.org/loop\"\n    icon = \"infinity\"\n\n    inputs = [\n        HandleInput(\n            name=\"data\",\n            display_name=\"Inputs\",\n            info=\"The initial DataFrame to iterate over.\",\n            input_types=[\"DataFrame\"],\n        ),\n    ]\n\n    outputs = [\n        Output(\n            display_name=\"Item\",\n            name=\"item\",\n            method=\"item_output\",\n            allows_loop=True,\n            loop_types=[\"Message\"],\n            group_outputs=True,\n        ),\n        Output(display_name=\"Done\", name=\"done\", method=\"done_output\", group_outputs=True),\n    ]\n\n    def initialize_data(self) -> None:\n        \"\"\"Initialize the data list, context index, and aggregated list.\"\"\"\n        if self.ctx.get(f\"{self._id}_initialized\", False):\n            return\n\n        # Ensure data is a list of Data objects\n        data_list = self._validate_data(self.data)\n\n        # Store the initial data and context variables\n        self.update_ctx(\n            {\n                f\"{self._id}_data\": data_list,\n                f\"{self._id}_index\": 0,\n                f\"{self._id}_aggregated\": [],\n                f\"{self._id}_initialized\": True,\n            }\n        )\n\n    def _convert_message_to_data(self, message: Message) -> Data:\n        \"\"\"Convert a Message object to a Data object using Type Convert logic.\"\"\"\n        return convert_to_data(message, auto_parse=False)\n\n    def _validate_data(self, data):\n        \"\"\"Validate and return a list of Data objects.\"\"\"\n        return validate_data_input(data)\n\n    def get_loop_body_vertices(self) -> set[str]:\n        \"\"\"Identify vertices in this loop's body via graph traversal.\n\n        Traverses from the loop's \"item\" output to the vertex that feeds back\n        to the loop's \"item\" input, collecting all vertices in between.\n        This naturally handles nested loops by stopping at this loop's feedback edge.\n\n        Returns:\n            Set of vertex IDs that form this loop's body\n        \"\"\"\n        # Check if we have a proper graph context\n        if not hasattr(self, \"_vertex\") or self._vertex is None:\n            return set()\n\n        return get_loop_body_vertices(\n            vertex=self._vertex,\n            graph=self.graph,\n            get_incoming_edge_by_target_param_fn=self.get_incoming_edge_by_target_param,\n        )\n\n    def _get_loop_body_start_vertex(self) -> str | None:\n        \"\"\"Get the first vertex in the loop body (connected to loop's item output).\n\n        Returns:\n            The vertex ID of the first vertex in the loop body, or None if not found\n        \"\"\"\n        # Check if we have a proper graph context\n        if not hasattr(self, \"_vertex\") or self._vertex is None:\n            return None\n\n        return get_loop_body_start_vertex(vertex=self._vertex)\n\n    def _extract_loop_output(self, results: list) -> Data:\n        \"\"\"Extract the output from subgraph execution results.\n\n        Args:\n            results: List of VertexBuildResult objects from subgraph execution\n\n        Returns:\n            Data object containing the loop iteration output\n        \"\"\"\n        # Get the vertex ID that feeds back to the item input (end of loop body)\n        end_vertex_id = self.get_incoming_edge_by_target_param(\"item\")\n        return extract_loop_output(results=results, end_vertex_id=end_vertex_id)\n\n    async def execute_loop_body(self, data_list: list[Data], event_manager=None) -> list[Data]:\n        \"\"\"Execute loop body for each data item.\n\n        Creates an isolated subgraph for the loop body and executes it\n        for each item in the data list, collecting results.\n\n        Args:\n            data_list: List of Data objects to iterate over\n            event_manager: Optional event manager to pass to subgraph execution for UI events\n\n        Returns:\n            List of Data objects containing results from each iteration\n        \"\"\"\n        # Get the loop body configuration once\n        loop_body_vertex_ids = self.get_loop_body_vertices()\n        start_vertex_id = self._get_loop_body_start_vertex()\n        start_edge = get_loop_body_start_edge(self._vertex)\n        end_vertex_id = self.get_incoming_edge_by_target_param(\"item\")\n\n        return await execute_loop_body(\n            graph=self.graph,\n            data_list=data_list,\n            loop_body_vertex_ids=loop_body_vertex_ids,\n            start_vertex_id=start_vertex_id,\n            start_edge=start_edge,\n            end_vertex_id=end_vertex_id,\n            event_manager=event_manager,\n        )\n\n    def item_output(self) -> Data:\n        \"\"\"Output is no longer used - loop executes internally now.\n\n        This method is kept for backward compatibility but does nothing.\n        The actual loop execution happens in done_output().\n        \"\"\"\n        self.stop(\"item\")\n        return Data(text=\"\")\n\n    async def done_output(self) -> DataFrame:\n        \"\"\"Execute the loop body for all items and return aggregated results.\n\n        This is now the main execution point for the loop. It:\n        1. Gets the data list to iterate over\n        2. Executes the loop body as an isolated subgraph for each item\n        3. Returns the aggregated results\n\n        Args:\n            event_manager: Optional event manager for UI event emission\n        \"\"\"\n        self.initialize_data()\n\n        # Get data list\n        data_list = self.ctx.get(f\"{self._id}_data\", [])\n\n        if not data_list:\n            return DataFrame([])\n\n        # Execute loop body for all items\n        try:\n            aggregated_results = await self.execute_loop_body(data_list, event_manager=self._event_manager)\n            return DataFrame(aggregated_results)\n        except Exception as e:\n            # Log error and return empty DataFrame\n            from lfx.log.logger import logger\n\n            await logger.aerror(f\"Error executing loop body: {e}\")\n            raise\n"
               },
               "data": {
                 "_input_type": "HandleInput",
@@ -1717,7 +1717,7 @@
                 "trace_as_metadata": true,
                 "track_in_telemetry": false,
                 "type": "str",
-                "value": "Text: {text}"
+                "value": "Title: {title}\nSummary: {summary}"
               },
               "sep": {
                 "_input_type": "MessageTextInput",