fix: guard event serialization against unserializable component payloads (#12591) (#13651)

* fix: guard event serialization against unserializable component payloads (#12591)

Loop flows containing a vector store (e.g. Chroma) failed to build with:

    Error building Component Loop:
    [TypeError("'_thread.lock' object is not iterable"),
     TypeError('vars() argument must have __dict__ attribute')]

The Loop runs its body in an isolated subgraph and streams an on_end_vertex
event for every completed inner vertex. EventManager.send_event serialized
that payload with a raw fastapi.jsonable_encoder call. When a payload carries
an object jsonable_encoder cannot serialize -- most notably a threading.Lock
held by a vector-DB client -- its last-resort fallback raises
ValueError([TypeError, TypeError]), which propagated out of the subgraph and
aborted the whole flow build.

The earlier fix (#12877) only guarded ResultData.validate_model and
build_output_logs (both already routed through the fail-safe serialize()), so
this jsonable_encoder path was never covered and the issue persisted.

send_event now falls back to lfx's fail-safe serialize() when jsonable_encoder
raises, degrading unknown objects to a string representation instead of
crashing. The happy path is unchanged (the fallback only runs on failure).

Adds a focused EventManager unit test and an end-to-end Loop regression test
whose body emits a lock-bearing Data (both fail without the fix).

* [autofix.ci] apply automated fixes

---------

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

src/lfx/src/lfx/events/event_manager.py

@@ -12,6 +12,7 @@ from fastapi.encoders import jsonable_encoder
 from typing_extensions import Protocol
 
 from lfx.log.logger import logger
+from lfx.serialization.serialization import serialize
 
 if TYPE_CHECKING:
     # Lightweight type stub for log types
@@ -69,7 +70,18 @@ class EventManager:
         self.events[name] = callback_
 
     def send_event(self, *, event_type: str, data: LoggableType):
-        jsonable_data = jsonable_encoder(data)
+        try:
+            jsonable_data = jsonable_encoder(data)
+        except (ValueError, TypeError) as exc:
+            # Component payloads can contain objects that FastAPI's jsonable_encoder
+            # cannot serialize — e.g. a vector-DB client holding a ``threading.Lock``
+            # surfaced through a Loop's per-iteration build events (issue #12591).
+            # jsonable_encoder's last-resort fallback raises ``ValueError([TypeError, ...])``
+            # there, which would otherwise abort the entire flow build. Degrade to the
+            # fail-safe serializer, which converts unknown objects to a string
+            # representation instead of crashing.
+            logger.debug(f"jsonable_encoder failed for event '{event_type}'; using safe serializer: {exc}")
+            jsonable_data = serialize(data, to_str=True)
         json_data = {"event": event_type, "data": jsonable_data}
         event_id = f"{event_type}-{uuid.uuid4()}"
         str_data = json.dumps(json_data) + "\n\n"

src/lfx/tests/unit/components/flow_controls/test_loop_unserializable_output.py

@@ -0,0 +1,121 @@
+"""Regression test for issue #12591: Loop + vector DB fails to build.
+
+A vector-store component inside a Loop body (e.g. Chroma) can surface objects
+that FastAPI's ``jsonable_encoder`` cannot serialize — most notably a
+``threading.Lock`` held by the underlying client. The Loop runs its body in an
+isolated subgraph and streams an ``on_end_vertex`` event for every completed
+inner vertex; that event payload was serialized with a raw ``jsonable_encoder``
+call, whose last-resort fallback raises
+``ValueError([TypeError("'_thread.lock' object is not iterable"),
+TypeError('vars() argument must have __dict__ attribute')])``.
+
+The exception propagated out of the subgraph and surfaced to the user as
+``Error building Component Loop: [TypeError(...), TypeError(...)]``, breaking the
+whole flow. This test drives a real Loop graph whose body emits a lock-bearing
+``Data`` and asserts the build completes instead of crashing.
+"""
+
+import asyncio
+import json
+import threading
+
+import pytest
+from lfx.components.flow_controls.loop import LoopComponent
+from lfx.custom.custom_component.component import Component
+from lfx.events.event_manager import create_default_event_manager
+from lfx.graph import Graph
+from lfx.io import HandleInput, Output
+from lfx.schema.data import Data
+from lfx.schema.dataframe import DataFrame
+
+
+class _LockEmittingComponent(Component):
+    """Loop-body node that emits a ``Data`` whose payload carries a lock.
+
+    Stands in for a vector-DB component (Chroma/Astra/…) whose output or
+    metadata references a client object holding a non-serializable
+    ``threading.Lock``.
+    """
+
+    display_name = "Lock Emitter"
+    name = "LockEmitter"
+
+    inputs = [HandleInput(name="input_value", display_name="In", input_types=["Data", "Message"])]
+    outputs = [Output(display_name="Out", name="out", method="build_out")]
+
+    def build_out(self) -> Data:
+        result = Data(data={"text": "result", "vector_store": threading.Lock()})
+        # Components surface their last output as ``status``; this is what feeds
+        # the build artifacts/events that get serialized for the UI.
+        self.status = result
+        return result
+
+
+def _attach_loop_feedback(loop: LoopComponent, source: Component) -> None:
+    """Wire ``source.out`` back into ``loop.item`` in the frontend edge shape.
+
+    Mirrors the helper used by the other loop graph tests: the loop feedback
+    edge has to be appended in the frontend-compatible shape and the source
+    component registered so ``Graph()`` walks to it.
+    """
+    loop._edges.append(
+        {
+            "source": source.get_id(),
+            "target": loop.get_id(),
+            "data": {
+                "sourceHandle": {
+                    "dataType": source.name,
+                    "id": source.get_id(),
+                    "name": "out",
+                    "output_types": ["Data"],
+                },
+                "targetHandle": {
+                    "dataType": "LoopComponent",
+                    "id": loop.get_id(),
+                    "name": "item",
+                    "output_types": ["Data", "Message"],
+                },
+            },
+        }
+    )
+    if source not in loop._components:
+        loop._components.append(source)
+
+
+@pytest.mark.asyncio
+async def test_loop_with_unserializable_body_output_builds():
+    """Loop body that emits a lock-bearing Data must not crash the build (#12591)."""
+    loop = LoopComponent(_id="loop")
+    rows = [Data(text=f"Row {i}") for i in range(3)]
+    loop.set(data=DataFrame(rows))
+
+    body = _LockEmittingComponent(_id="LockEmitter-body")
+    body.set(input_value=loop.item_output)
+    _attach_loop_feedback(loop, body)
+
+    # A downstream consumer on `done` so done_output runs (classic Loop topology).
+    done_sink = LoopComponent(_id="loop-done-sink")
+    done_sink.set(data=loop.done_output)
+
+    graph = Graph(loop, done_sink)
+    queue: asyncio.Queue = asyncio.Queue()
+    event_manager = create_default_event_manager(queue=queue)
+
+    # Pre-fix this raised ComponentBuildError("Error building Component Loop: "
+    # "[TypeError(\"'_thread.lock' object is not iterable\"), ...]").
+    results = [result async for result in graph.async_start(event_manager=event_manager)]
+
+    loop_result = next(r for r in results if getattr(r, "vertex", None) and r.vertex.id == "loop")
+    assert loop_result.valid, "Loop vertex should build successfully despite unserializable body output"
+
+    # The inner body vertex's end event was emitted (not dropped) once per row,
+    # with the unserializable lock degraded to a string instead of crashing.
+    end_vertex_events = []
+    while not queue.empty():
+        _, raw, _ = await queue.get()
+        payload = json.loads(raw.decode("utf-8").strip())
+        if payload["event"] == "end_vertex":
+            end_vertex_events.append(payload["data"])
+
+    body_runs = [e for e in end_vertex_events if e.get("build_data", {}).get("id") == "LockEmitter-body"]
+    assert len(body_runs) == len(rows), f"Expected {len(rows)} body iterations, got {len(body_runs)}"

src/lfx/tests/unit/events/test_event_manager.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import json
+import threading
 from unittest.mock import MagicMock
 
 import pytest
@@ -130,6 +131,37 @@ class TestEventManager:
         # Should not raise exception, just log debug message
         manager.send_event(event_type="test", data=test_data)
 
+    def test_send_event_degrades_unserializable_payload(self):
+        """Regression for #12591: an unserializable payload must not crash the build.
+
+        A ``threading.Lock`` held by a vector-DB client (surfaced through a Loop's
+        per-iteration build events) makes FastAPI's jsonable_encoder raise
+        ``ValueError([TypeError, TypeError])``. ``send_event`` should fall back to the
+        fail-safe serializer and still emit a JSON-decodable event with the object
+        degraded to a string.
+        """
+        queue = MagicMock()
+        manager = EventManager(queue)
+
+        lock = threading.Lock()
+        # Mirror the on_end_vertex build_data shape emitted during a loop subgraph build.
+        data = {"build_data": {"id": "Chroma-1", "data": {"text": "chunk", "vector_store": lock}}}
+
+        # Must not raise (pre-fix this propagated as "Error building Component Loop").
+        manager.send_event(event_type="end_vertex", data=data)
+
+        queue.put_nowait.assert_called_once()
+        _, data_bytes, _ = queue.put_nowait.call_args[0][0]
+        parsed = json.loads(data_bytes.decode("utf-8").strip())
+
+        assert parsed["event"] == "end_vertex"
+        # The lock survives as a string representation rather than crashing serialization.
+        serialized_lock = parsed["data"]["build_data"]["data"]["vector_store"]
+        assert isinstance(serialized_lock, str)
+        assert "lock" in serialized_lock.lower()
+        # Serializable siblings are preserved.
+        assert parsed["data"]["build_data"]["data"]["text"] == "chunk"
+
     def test_noop_method(self):
         """Test noop method."""
         queue = asyncio.Queue()