fix(AIC-3574): build the LaunchDarkly client off the event loop

packages/client/src/launchdarkly_ai_server/lifecycle.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import importlib
 import inspect
 import logging
@@ -301,7 +302,30 @@ async def init_client(
 
     ld_config = config_cls(sdk_key, **config_kwargs)
     # start_wait caps the blocking init time; matches the TS SDK's 10 s timeout.
-    ld_client = client_cls(ld_config, start_wait=10)
+    # Built in a thread so that wait does not stall the caller's event loop.
+    build = asyncio.ensure_future(
+        asyncio.to_thread(client_cls, ld_config, start_wait=10)
+    )
+    try:
+        ld_client = await asyncio.shield(build)
+    except asyncio.CancelledError:
+        # The thread can't be stopped: close the client it finishes building, or its
+        # streaming connection outlives this cancelled call.
+        build.add_done_callback(_close_abandoned_client)
+        raise
+
+    # A concurrent first call may have won while this one was connecting: keep
+    # its client and close ours, so only one streaming connection survives.
+    if _client is not None:
+        try:
+            close_result = ld_client.close()
+            if inspect.isawaitable(close_result):
+                await close_result
+        except Exception:
+            pass
+        flush_ai_sdk_info(_client)
+        flush_sdk_usage(_client)
+        return _client
 
     # As on the BYOC path: only a fully set-up client becomes the singleton. We
     # built this one, so close it on failure — it holds a streaming connection
@@ -322,6 +346,24 @@ async def init_client(
     return _client
 
 
+def _close_abandoned_client(build: asyncio.Future[Any]) -> None:
+    """Closes a client whose ``init_client`` call was cancelled while it was being built."""
+    if build.cancelled() or build.exception() is not None:
+        return
+    try:
+        close_result = build.result().close()
+        if inspect.isawaitable(close_result):
+            # The loop keeps only a weak reference to the task; hold it until it is done.
+            task = asyncio.ensure_future(close_result)
+            _closing_tasks.add(task)
+            task.add_done_callback(_closing_tasks.discard)
+    except Exception:
+        pass
+
+
+_closing_tasks: set[asyncio.Future[Any]] = set()
+
+
 def _release_otel_globals() -> None:
     """
     Releases the process-global tracer provider that ``_setup_telemetry``

packages/client/tests/test_lifecycle.py

@@ -3,9 +3,11 @@
 Reference: TESTING.md §3.9
 """
 
+import asyncio
 import os
 import subprocess
 import sys
+import time
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -326,6 +328,69 @@ async def test_returns_initialized_client(self) -> None:
                     result = await init_client()
         assert result is stub
 
+    async def test_builds_the_sdk_client_off_the_event_loop(self) -> None:
+        # LDClient blocks up to start_wait (10 s) while it connects; built on the
+        # loop, that stalls every other task in the app.
+        stub = _make_stub_client()
+        mock_ld = MagicMock()
+        mock_ld.Config = MagicMock(return_value=MagicMock())
+        mock_ld.LDClient = MagicMock(
+            side_effect=lambda *_a, **_k: (time.sleep(0.3), stub)[1]
+        )
+        ticks = 0
+
+        async def tick() -> None:
+            nonlocal ticks
+            while True:
+                await asyncio.sleep(0.01)
+                ticks += 1
+
+        ticker = asyncio.create_task(tick())
+        with patch.dict(os.environ, {"LD_SDK_KEY": "test-key"}):
+            with patch.object(lifecycle_module, "_setup_telemetry", return_value=None):
+                with patch("importlib.import_module", return_value=mock_ld):
+                    assert await init_client() is stub
+        ticker.cancel()
+        assert ticks >= 5
+
+    async def test_concurrent_first_calls_end_with_one_client(self) -> None:
+        first, second = _make_stub_client(), _make_stub_client()
+        built = iter([first, second])
+        mock_ld = MagicMock()
+        mock_ld.Config = MagicMock(return_value=MagicMock())
+        mock_ld.LDClient = MagicMock(
+            side_effect=lambda *_a, **_k: (time.sleep(0.05), next(built))[1]
+        )
+        with patch.dict(os.environ, {"LD_SDK_KEY": "test-key"}):
+            with patch.object(lifecycle_module, "_setup_telemetry", return_value=None):
+                with patch("importlib.import_module", return_value=mock_ld):
+                    a, b = await asyncio.gather(init_client(), init_client())
+        assert a is b is get_client()
+        extra = second if a is first else first
+        extra.close.assert_awaited_once()
+        a.close.assert_not_called()
+
+    async def test_cancelled_call_closes_the_client_still_being_built(self) -> None:
+        # The thread can't be stopped, so the client it finishes building after the
+        # cancel holds a streaming connection that nothing else would close.
+        stub = _make_stub_client()
+        mock_ld = MagicMock()
+        mock_ld.Config = MagicMock(return_value=MagicMock())
+        mock_ld.LDClient = MagicMock(
+            side_effect=lambda *_a, **_k: (time.sleep(0.2), stub)[1]
+        )
+        with patch.dict(os.environ, {"LD_SDK_KEY": "test-key"}):
+            with patch.object(lifecycle_module, "_setup_telemetry", return_value=None):
+                with patch("importlib.import_module", return_value=mock_ld):
+                    task = asyncio.create_task(init_client())
+                    await asyncio.sleep(0.05)
+                    task.cancel()
+                    with pytest.raises(asyncio.CancelledError):
+                        await task
+                    await asyncio.sleep(0.4)
+        stub.close.assert_awaited_once()
+        assert lifecycle_module._client is None
+
 
 # ---------------------------------------------------------------------------
 # OTel degradation