fix(tracing): detach orphan spans before database flush (#14243)

fix(tracing): detach spans from missing parents

Author: erichare

src/backend/base/langflow/services/tracing/span_sorting.py

@@ -63,22 +63,37 @@ def topological_sort_spans(
     sort over the batch so that every parent is inserted first.
 
     Spans whose ``parent_span_id`` points outside the current batch are
-    treated as roots (the parent already exists in the DB from a prior
-    flush).
+    detached from that missing parent. Native tracing persists a trace's
+    spans in a single batch, so retaining the reference would violate the
+    immediate foreign-key constraint.
     """
     batch_ids = {span_uuid for _, span_uuid, _ in resolved}
 
+    normalized: list[tuple[dict[str, Any], UUID, UUID | None]] = []
+    missing_parent_count = 0
+    for span_data, span_uuid, parent_uuid in resolved:
+        normalized_parent_uuid = parent_uuid
+        if normalized_parent_uuid is not None and normalized_parent_uuid not in batch_ids:
+            normalized_parent_uuid = None
+            missing_parent_count += 1
+        normalized.append((span_data, span_uuid, normalized_parent_uuid))
+
+    if missing_parent_count:
+        logger.warning(
+            "Detached %d tracing spans from missing parents to preserve database integrity.",
+            missing_parent_count,
+        )
+
     sorted_spans: list[tuple[dict[str, Any], UUID, UUID | None]] = []
     inserted: set[UUID] = set()
-    remaining = list(resolved)
+    remaining = normalized
 
     while remaining:
         next_round: list[tuple[dict[str, Any], UUID, UUID | None]] = []
         progress = False
         for item in remaining:
             _, span_uuid, parent_uuid = item
-            # Insert if: no parent, parent outside batch, or parent already inserted
-            if parent_uuid is None or parent_uuid not in batch_ids or parent_uuid in inserted:
+            if parent_uuid is None or parent_uuid in inserted:
                 sorted_spans.append(item)
                 inserted.add(span_uuid)
                 progress = True

src/backend/tests/unit/services/tracing/test_native_tracer.py

@@ -691,13 +691,14 @@ def test_deep_nesting(self):
         assert uuids.index(root) < uuids.index(mid) < uuids.index(leaf)
 
     def test_parent_outside_batch(self):
-        """Spans referencing a parent not in the batch are treated as roots."""
+        """Spans referencing a missing parent are detached before insertion."""
         external_parent = uuid4()
         child_id = uuid4()
         items = [self._make_span(child_id, external_parent)]
         result = topological_sort_spans(items)
         assert len(result) == 1
         assert result[0][1] == child_id
+        assert result[0][2] is None
 
     def test_mixed_roots_and_children(self):
         root_a = uuid4()
@@ -891,3 +892,43 @@ async def capture_merge(obj):
         assert span_objects[0].id == parent_uuid
         assert span_objects[1].id == child_uuid
         assert span_objects[1].parent_span_id == parent_uuid
+
+    async def test_flush_detaches_span_from_missing_parent(self):
+        """A missing parent must not leave an invalid self-referential FK."""
+        tracer = _make_tracer(flow_id=str(uuid4()))
+        tracer.completed_spans = [
+            {
+                "id": str(uuid4()),
+                "name": "Orphan Span",
+                "span_type": SpanType.CHAIN,
+                "inputs": {},
+                "outputs": None,
+                "start_time": datetime.now(tz=timezone.utc),
+                "end_time": datetime.now(tz=timezone.utc),
+                "latency_ms": 5,
+                "status": SpanStatus.OK,
+                "error": None,
+                "attributes": {},
+                "span_source": "langchain",
+                "parent_span_id": uuid4(),
+            }
+        ]
+
+        merged_objects = []
+        mock_session = AsyncMock()
+        mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+        mock_session.__aexit__ = AsyncMock(return_value=False)
+
+        async def capture_merge(obj):
+            merged_objects.append(obj)
+
+        mock_session.merge = capture_merge
+
+        with patch("lfx.services.deps.session_scope", return_value=mock_session):
+            await tracer._flush_to_database()
+
+        from langflow.services.database.models.traces.model import SpanTable
+
+        span_objects = [obj for obj in merged_objects if isinstance(obj, SpanTable)]
+        assert len(span_objects) == 1
+        assert span_objects[0].parent_span_id is None