perf(http): pool remaining raw aiohttp.ClientSession sites onto shared client (#12979)

autobot-backend/api/realtime_session_test.py

@@ -53,7 +53,13 @@ def _form(sdp: str = _SDP_OFFER, session: str = _SESSION_JSON) -> dict:
 
 
 def _mock_upstream(status: int = 200, body: bytes = _SDP_ANSWER):
-    """Return a mock aiohttp response context-manager."""
+    """Return a mock aiohttp response, itself the tracked_request() async CM.
+
+    #12979: OpenAIRealtimeProvider._post() now routes through the shared
+    pool's get_http_client().tracked_request() rather than a private
+    ClientSession, so this is patched directly (see _patch_http_client())
+    instead of patching aiohttp.ClientSession.
+    """
     resp = MagicMock()
     resp.status = status
     resp.read = AsyncMock(return_value=body)
@@ -64,15 +70,11 @@ def _mock_upstream(status: int = 200, body: bytes = _SDP_ANSWER):
     return cm
 
 
-def _mock_session(upstream_cm):
-    """Return a mock aiohttp.ClientSession context-manager."""
-    session = MagicMock()
-    session.post = MagicMock(return_value=upstream_cm)
+def _patch_http_client(upstream_cm):
+    """Patch the shared HTTPClientManager singleton's tracked_request()."""
+    from autobot_shared.http_client import get_http_client
 
-    session_cm = MagicMock()
-    session_cm.__aenter__ = AsyncMock(return_value=session)
-    session_cm.__aexit__ = AsyncMock(return_value=False)
-    return session_cm
+    return patch.object(get_http_client(), "tracked_request", return_value=upstream_cm)
 
 
 # ---------------------------------------------------------------------------
@@ -85,13 +87,13 @@ class TestHappyPath:
 
     def test_returns_200(self, client: TestClient):
         upstream_cm = _mock_upstream(200, _SDP_ANSWER)
-        session_cm = _mock_session(upstream_cm)
+        http_client_patch = _patch_http_client(upstream_cm)
 
         with (
             patch(
                 "voice_processing.realtime.openai_provider.OpenAIRealtimeProvider._api_key", return_value="sk-test-key"
             ),
-            patch("aiohttp.ClientSession", return_value=session_cm),
+            http_client_patch,
         ):
             resp = client.post(
                 "/api/voice/realtime/session",
@@ -102,13 +104,13 @@ def test_returns_200(self, client: TestClient):
 
     def test_returns_sdp_content_type(self, client: TestClient):
         upstream_cm = _mock_upstream(200, _SDP_ANSWER)
-        session_cm = _mock_session(upstream_cm)
+        http_client_patch = _patch_http_client(upstream_cm)
 
         with (
             patch(
                 "voice_processing.realtime.openai_provider.OpenAIRealtimeProvider._api_key", return_value="sk-test-key"
             ),
-            patch("aiohttp.ClientSession", return_value=session_cm),
+            http_client_patch,
         ):
             resp = client.post(
                 "/api/voice/realtime/session",
@@ -119,13 +121,13 @@ def test_returns_sdp_content_type(self, client: TestClient):
 
     def test_returns_upstream_body(self, client: TestClient):
         upstream_cm = _mock_upstream(200, _SDP_ANSWER)
-        session_cm = _mock_session(upstream_cm)
+        http_client_patch = _patch_http_client(upstream_cm)
 
         with (
             patch(
                 "voice_processing.realtime.openai_provider.OpenAIRealtimeProvider._api_key", return_value="sk-test-key"
             ),
-            patch("aiohttp.ClientSession", return_value=session_cm),
+            http_client_patch,
         ):
             resp = client.post(
                 "/api/voice/realtime/session",
@@ -136,13 +138,13 @@ def test_returns_upstream_body(self, client: TestClient):
 
     def test_accepts_201_from_upstream(self, client: TestClient):
         upstream_cm = _mock_upstream(201, _SDP_ANSWER)
-        session_cm = _mock_session(upstream_cm)
+        http_client_patch = _patch_http_client(upstream_cm)
 
         with (
             patch(
                 "voice_processing.realtime.openai_provider.OpenAIRealtimeProvider._api_key", return_value="sk-test-key"
             ),
-            patch("aiohttp.ClientSession", return_value=session_cm),
+            http_client_patch,
         ):
             resp = client.post(
                 "/api/voice/realtime/session",
@@ -190,13 +192,13 @@ class TestUpstream401:
 
     def test_upstream_401_returns_502(self, client: TestClient):
         upstream_cm = _mock_upstream(401, b'{"error":"unauthorized"}')
-        session_cm = _mock_session(upstream_cm)
+        http_client_patch = _patch_http_client(upstream_cm)
 
         with (
             patch(
                 "voice_processing.realtime.openai_provider.OpenAIRealtimeProvider._api_key", return_value="sk-bad-key"
             ),
-            patch("aiohttp.ClientSession", return_value=session_cm),
+            http_client_patch,
         ):
             resp = client.post(
                 "/api/voice/realtime/session",
@@ -207,13 +209,13 @@ def test_upstream_401_returns_502(self, client: TestClient):
 
     def test_upstream_401_error_body(self, client: TestClient):
         upstream_cm = _mock_upstream(401, b'{"error":"unauthorized"}')
-        session_cm = _mock_session(upstream_cm)
+        http_client_patch = _patch_http_client(upstream_cm)
 
         with (
             patch(
                 "voice_processing.realtime.openai_provider.OpenAIRealtimeProvider._api_key", return_value="sk-bad-key"
             ),
-            patch("aiohttp.ClientSession", return_value=session_cm),
+            http_client_patch,
         ):
             resp = client.post(
                 "/api/voice/realtime/session",
@@ -235,13 +237,13 @@ class TestUpstream5xx:
     @pytest.mark.parametrize("status", [500, 502, 503, 504])
     def test_upstream_5xx_returns_502(self, client: TestClient, status: int):
         upstream_cm = _mock_upstream(status, b"internal server error")
-        session_cm = _mock_session(upstream_cm)
+        http_client_patch = _patch_http_client(upstream_cm)
 
         with (
             patch(
                 "voice_processing.realtime.openai_provider.OpenAIRealtimeProvider._api_key", return_value="sk-test-key"
             ),
-            patch("aiohttp.ClientSession", return_value=session_cm),
+            http_client_patch,
         ):
             resp = client.post(
                 "/api/voice/realtime/session",
@@ -252,13 +254,13 @@ def test_upstream_5xx_returns_502(self, client: TestClient, status: int):
 
     def test_upstream_5xx_error_body(self, client: TestClient):
         upstream_cm = _mock_upstream(500, b"oops")
-        session_cm = _mock_session(upstream_cm)
+        http_client_patch = _patch_http_client(upstream_cm)
 
         with (
             patch(
                 "voice_processing.realtime.openai_provider.OpenAIRealtimeProvider._api_key", return_value="sk-test-key"
             ),
-            patch("aiohttp.ClientSession", return_value=session_cm),
+            http_client_patch,
         ):
             resp = client.post(
                 "/api/voice/realtime/session",
@@ -279,13 +281,13 @@ class TestProviderDispatch:
 
     def test_session_header_advertises_provider(self, client: TestClient):
         upstream_cm = _mock_upstream(200, _SDP_ANSWER)
-        session_cm = _mock_session(upstream_cm)
+        http_client_patch = _patch_http_client(upstream_cm)
         with (
             patch(
                 "voice_processing.realtime.openai_provider.OpenAIRealtimeProvider._api_key",
                 return_value="sk-test-key",
             ),
-            patch("aiohttp.ClientSession", return_value=session_cm),
+            http_client_patch,
         ):
             resp = client.post(
                 "/api/voice/realtime/session",

autobot-backend/llm_shared/mock_providers.py

@@ -131,24 +131,27 @@ async def generate(self, prompt: str, model: str | None = None) -> dict:
         try:
             import aiohttp
 
+            from autobot_shared.http_client import get_http_client
+
             data = {
                 "model": model or self._default_model,
                 "messages": [{"role": "user", "content": prompt}],
                 "stream": False,
             }
 
-            async with aiohttp.ClientSession() as session:
-                timeout = aiohttp.ClientTimeout(total=TIMEOUT_HTTP_LONG)
-                async with session.post(f"{self._ollama_url}/api/chat", json=data, timeout=timeout) as response:
-                    if response.status != 200:
-                        error_text = await response.text()
-                        logger.error(
-                            "Ollama request failed: HTTP %s - %s",
-                            response.status,
-                            error_text,
-                        )
-                        return self._create_error_response(f"Error: Ollama returned HTTP {response.status}")
-                    result = await response.json()
+            timeout = aiohttp.ClientTimeout(total=TIMEOUT_HTTP_LONG)
+            async with get_http_client().tracked_request(
+                "POST", f"{self._ollama_url}/api/chat", json=data, timeout=timeout
+            ) as response:
+                if response.status != 200:
+                    error_text = await response.text()
+                    logger.error(
+                        "Ollama request failed: HTTP %s - %s",
+                        response.status,
+                        error_text,
+                    )
+                    return self._create_error_response(f"Error: Ollama returned HTTP {response.status}")
+                result = await response.json()
 
             return self._format_ollama_response(result)
 

autobot-backend/media/link/pipeline.py

@@ -222,12 +222,15 @@ async def _fetch_and_parse(self, url: str, metadata: Dict) -> Dict[str, Any]:
         # cert verification for known-safe internal URLs.
         ssl_context = False if metadata.get("allow_self_signed") else None
         try:
-            async with aiohttp.ClientSession(headers=headers, timeout=_DEFAULT_TIMEOUT) as session:
-                async with session.get(url, allow_redirects=True, ssl=ssl_context) as response:
-                    final_url = str(response.url)
-                    content_type = response.headers.get("Content-Type", "")
-                    raw_html = await response.text(encoding="utf-8", errors="replace")
-                    status = response.status
+            from autobot_shared.http_client import get_http_client
+
+            async with get_http_client().tracked_request(
+                "GET", url, headers=headers, timeout=_DEFAULT_TIMEOUT, allow_redirects=True, ssl=ssl_context
+            ) as response:
+                final_url = str(response.url)
+                content_type = response.headers.get("Content-Type", "")
+                raw_html = await response.text(encoding="utf-8", errors="replace")
+                status = response.status
 
             if status >= 400:
                 return self._error_result(