fix: instrument the database engine, not the global dispatcher

The DB spans this PR added did not deliver what it claimed. Verified against
a live backend: one API request produced 13 sqlalchemy connect spans and zero
query spans, with no db.statement anywhere. Connection-pool noise dressed up
as database visibility.

Langflow builds an AsyncEngine, and the instrumentor patches the sync engine
underneath it. instrument_dependencies() was called from the app factory with
no engine, because the engine does not exist that early, so SQLAlchemy
instrumented globally and attached to pool events only.

Instrument where the engine is built instead, passing it. Renamed to
instrument_database, and the engine argument is required rather than optional:
the engine-less call is exactly the broken configuration, and a global
instrument() would also win the race and make a later engine-specific call a
silent no-op.

Same request after the change: 28 query spans carrying db.statement, bound
parameters still placeholders, and the flow's input text absent from the trace.

The unit tests passed throughout. They assert the scope is allowlisted and the
instrumentor is wired, not that a query span reaches an exporter, so this was
only visible against a real backend.

Author: ogabrielluiz

.secrets.baseline

@@ -6688,7 +6688,7 @@
         "filename": "src/lfx/src/lfx/cli/serve_app.py",
         "hashed_secret": "b894b81be94cf8fa8d7536475aaec876addf05c8",
         "is_verified": false,
-        "line_number": 71,
+        "line_number": 70,
         "is_secret": false
       }
     ],

src/backend/base/langflow/main.py

@@ -22,7 +22,6 @@
 from lfx.interface.utils import setup_llm_caching
 from lfx.log.logger import configure, logger
 from lfx.observability import (
-    instrument_dependencies,
     instrument_fastapi_app,
     start_event_loop_lag_monitor,
     stop_event_loop_lag_monitor,
@@ -993,9 +992,6 @@ async def exception_handler(_request: Request, exc: Exception):
     # lazy-include route patch + instrument_app). The helper lives in lfx so lfx serve
     # instruments its own app the same way.
     instrument_fastapi_app(app)
-    # The rest of the trace tree: DB queries and outbound HTTP made during a run. The engine is
-    # not built yet at app-factory time, so SQLAlchemy instruments its own dispatcher globally.
-    instrument_dependencies()
 
     add_pagination(app)
 

src/backend/base/langflow/services/database/service.py

@@ -16,6 +16,7 @@
 from alembic import command, util
 from alembic.config import Config
 from lfx.log.logger import logger
+from lfx.observability import instrument_database
 from lfx.services.deps import session_scope
 from sqlalchemy import event, inspect
 from sqlalchemy.dialects import sqlite as dialect_sqlite
@@ -280,6 +281,12 @@ def __init__(self, settings_service: SettingsService):
             self.engine = self._create_engine_with_retry()
         else:
             self.engine = self._create_engine()
+        # Here rather than in the app factory: this is an AsyncEngine, and the instrumentor
+        # patches the sync engine underneath it. Instrumenting globally instead (no engine
+        # argument) attaches to pool events only, which yields a connect span per checkout and
+        # not a single query span. Verified against a live backend: 13 connect spans and zero
+        # db.statement for one API request.
+        instrument_database(self.engine)
 
         # Create async session maker for efficient session creation
         # This is the recommended SQLAlchemy 2.0+ pattern

src/lfx/src/lfx/cli/serve_app.py

@@ -47,7 +47,6 @@
 from lfx.log.logger import logger
 from lfx.observability import (
     bootstrap_application_telemetry,
-    instrument_dependencies,
     instrument_fastapi_app,
     start_event_loop_lag_monitor,
     stop_event_loop_lag_monitor,
@@ -708,7 +707,6 @@ async def _lifespan(_app: FastAPI):
     )
 
     instrument_fastapi_app(app)
-    instrument_dependencies()
 
     app.state.registry = registry
     # Snapshot the API key once so per-request auth (verify_api_key, run on a threadpool

src/lfx/src/lfx/observability.py

@@ -627,37 +627,37 @@ def bootstrap_application_telemetry(*, prometheus_enabled: bool = False) -> Appl
     )
 
 
-def instrument_dependencies(*, engine: object | None = None) -> None:
-    """Instrument the database, so a slow request can be attributed to the queries it made.
+def instrument_database(engine: object) -> None:
+    """Instrument a SQLAlchemy engine, so a slow request can be attributed to its queries.
 
-    Deliberately does NOT instrument httpx or requests. Those are the transports the LLM
-    vendor SDKs ride on, and instrumenting them globally would put one span per outbound
-    provider call into the operator's APM. That boundary is held elsewhere too (the export
-    filter does not allowlist those scopes), and outbound provider health is delivered as
-    leak-safe metrics rather than spans.
+    Takes the engine rather than instrumenting globally. Langflow's engine is an AsyncEngine and
+    the instrumentor patches the sync engine underneath it; a global ``instrument()`` with no
+    engine attaches to pool events only, which produces a connect span per checkout and no query
+    spans at all. Verified against a live backend: 13 connect spans and zero db.statement for one
+    API request. That is worse than nothing, because it looks like DB visibility.
 
-    The instrumentor is given an explicit ``tracer_provider``. That is not decoration: a bare
-    ``instrument()`` binds to whatever provider is global, which is how vendor SDKs end up
-    exporting through ours.
+    Deliberately does not instrument httpx or requests. Those are the transports the LLM vendor
+    SDKs ride on, and instrumenting them globally would put one span per outbound provider call
+    into the operator's APM.
 
     Optional and failure-tolerant: a missing package or a double-instrument call must not take
     the app down, because none of this is worth a failed boot.
     """
     _instrument_sqlalchemy(engine)
 
 
-def _instrument_sqlalchemy(engine: object | None) -> None:
+def _instrument_sqlalchemy(engine: object) -> None:
     try:
         from opentelemetry import trace
         from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
     except ImportError:
         return
     try:
-        kwargs = {"tracer_provider": trace.get_tracer_provider()}
-        if engine is not None:
-            # An async engine exposes the sync one the instrumentor actually patches.
-            kwargs["engine"] = getattr(engine, "sync_engine", engine)
-        SQLAlchemyInstrumentor().instrument(**kwargs)
+        SQLAlchemyInstrumentor().instrument(
+            # An AsyncEngine exposes the sync one the instrumentor actually patches.
+            engine=getattr(engine, "sync_engine", engine),
+            tracer_provider=trace.get_tracer_provider(),
+        )
     except Exception:  # noqa: BLE001 - see above
         logger.debug("sqlalchemy instrumentation unavailable; DB spans will be missing", exc_info=True)