test: prove H2 bulkheads on the wire; lock sync lazy init

Add TLS+ALPN integration tests that saturate waits / stall upload bodies
and assert create lands on a different connection and stays fast. Guard
sync bulkhead client creation with a lock and enumerate long-lived
resource paths so new endpoints cannot silently join the API pool.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: Reflex

src/runloop_api_client/_base_client.py

@@ -1073,6 +1073,7 @@ def __init__(
         self._closed = False
         self._background_clients = {}
         self._transfer_clients = {}
+        self._bulkhead_lock = threading.Lock()
         self._background_pool_shards = background_pool_shards
         self._transfer_pool_shards = transfer_pool_shards
         # Custom http_client owns the full transport stack; don't invent sibling pools.
@@ -1118,29 +1119,37 @@ def _ensure_background_client(self, shard: int) -> httpx.Client:
         existing = self._background_clients.get(shard)
         if existing is not None:
             return existing
-        if self._uses_shared_pool:
-            transport: httpx.BaseTransport | None = _acquire_shared_sync_transport(
-                _shared_sync_background_transports, shard
-            )
-        else:
-            transport = None
-        client = self._make_bulkhead_client(transport=transport)
-        self._background_clients[shard] = client
-        return client
+        with self._bulkhead_lock:
+            existing = self._background_clients.get(shard)
+            if existing is not None:
+                return existing
+            if self._uses_shared_pool:
+                transport: httpx.BaseTransport | None = _acquire_shared_sync_transport(
+                    _shared_sync_background_transports, shard
+                )
+            else:
+                transport = None
+            client = self._make_bulkhead_client(transport=transport)
+            self._background_clients[shard] = client
+            return client
 
     def _ensure_transfer_client(self, shard: int) -> httpx.Client:
         existing = self._transfer_clients.get(shard)
         if existing is not None:
             return existing
-        if self._uses_shared_pool:
-            transport: httpx.BaseTransport | None = _acquire_shared_sync_transport(
-                _shared_sync_transfer_transports, shard
-            )
-        else:
-            transport = None
-        client = self._make_bulkhead_client(transport=transport)
-        self._transfer_clients[shard] = client
-        return client
+        with self._bulkhead_lock:
+            existing = self._transfer_clients.get(shard)
+            if existing is not None:
+                return existing
+            if self._uses_shared_pool:
+                transport: httpx.BaseTransport | None = _acquire_shared_sync_transport(
+                    _shared_sync_transfer_transports, shard
+                )
+            else:
+                transport = None
+            client = self._make_bulkhead_client(transport=transport)
+            self._transfer_clients[shard] = client
+            return client
 
     def _send_client_for_request(self, request: httpx.Request) -> httpx.Client:
         if not self._isolate_workload_pools: