fix(langfuse): re-check holders after the registry snapshot when reaping orphans

A sweep waiting on the registry lock held a pre-registration holder snapshot, so a
rotation racing in between could retire the provider of a just-returned client.
Holders are now snapshotted last and provider tracking shares the sweep's lock.

litellm/integrations/langfuse/langfuse_sdk.py

@@ -191,7 +191,8 @@ def build_isolated_tracer_provider(*, environment: str | None, release: str | No
         resource=Resource.create(dict(attributes)),
         sampler=TraceIdRatioBased(sample_rate) if sample_rate < 1 else None,
     )
-    _litellm_built_providers.add(provider)
+    with _LIVE_CLIENTS_LOCK:
+        _litellm_built_providers.add(provider)
     return provider
 
 
@@ -242,23 +243,25 @@ def _retire_orphaned_providers() -> None:
     prompt-management LRU drops clients, never reaches ``shutdown_langfuse_client``, and the
     provider's own atexit hook would keep its export thread alive for the rest of the process.
 
-    Providers are snapshotted before the registry: a provider is only ever built and its
-    bundle registered inside one registry-locked block, so any provider seen in the first
-    snapshot is fully registered by the time the second one is taken. Runs outside both
-    locks because provider shutdown flushes and joins the export thread.
+    Holders are snapshotted last: a client is registered in the same registry-locked block
+    that builds its provider, so once the registry snapshot's lock has been acquired, the
+    client of any provider from the first snapshot is visible to the final one even when a
+    concurrent rotation already evicted its bundle again. Runs outside both locks because
+    provider shutdown flushes and joins the export thread.
     """
     with _LIVE_CLIENTS_LOCK:
         candidates: Final = tuple(_litellm_built_providers)
-        held: Final = tuple(
-            getattr(resources, "tracer_provider", None)
-            for resources, holders in _live_clients.items()
-            if len(holders) > 0
-        )
     with LangfuseResourceManager._lock:  # pyright: ignore[reportPrivateUsage]  # registry has no public accessor
         registered: Final = tuple(
             getattr(resources, "tracer_provider", None)
             for resources in LangfuseResourceManager._instances.values()  # pyright: ignore[reportPrivateUsage]  # registry has no public accessor
         )
+    with _LIVE_CLIENTS_LOCK:
+        held: Final = tuple(
+            getattr(resources, "tracer_provider", None)
+            for resources, holders in _live_clients.items()
+            if len(holders) > 0
+        )
     orphaned: Final = tuple(provider for provider in candidates if provider not in registered and provider not in held)
     for provider in orphaned:
         _litellm_built_providers.discard(provider)

tests/test_litellm/integrations/langfuse/test_langfuse_sdk.py

@@ -649,6 +649,40 @@ def test_the_registrys_current_bundle_is_not_reaped_when_its_clients_die():
     assert _exports(successor, exporter, "after-collection")
 
 
+def test_a_sweep_overlapping_registration_and_rotation_keeps_the_live_provider():
+    """A sweep can snapshot providers before a client registers, then wait on the registry
+    lock while that client registers and a rotation evicts its fresh bundle. Holders are
+    re-read after the registry snapshot, so the stale first look must not win."""
+    import threading
+
+    from litellm.integrations.langfuse.langfuse_sdk import _retire_orphaned_providers
+
+    exporter = InMemorySpanExporter()
+    provider = build_isolated_tracer_provider(environment=None, release=None)
+    provider.add_span_processor(SimpleSpanProcessor(exporter))
+    client = Langfuse(
+        public_key=PUBLIC_KEY,
+        secret_key="sk-original",
+        host="http://127.0.0.1:1",
+        tracer_provider=provider,
+        span_exporter=exporter,
+    )
+    registry_lock = LangfuseResourceManager._lock
+    registry_lock.acquire()
+    try:
+        sweeper = threading.Thread(target=_retire_orphaned_providers)
+        sweeper.start()
+        sweeper.join(timeout=0.5)  # parks on the registry lock once its provider snapshot is taken
+        register_langfuse_client(client)
+        LangfuseResourceManager._instances.pop(PUBLIC_KEY, None)  # the rotation that evicts the fresh bundle
+    finally:
+        registry_lock.release()
+    sweeper.join(timeout=5)
+    assert not sweeper.is_alive()
+
+    assert _exports(client, exporter, "after-racing-sweep")
+
+
 def test_ssl_exporter_is_only_built_with_custom_tls_material(monkeypatch, tmp_path):
     """v4 exports over its own OTLP channel, so litellm's CA bundle must be rebuilt onto it."""
     import litellm