perf(http): route integrations/base+github+m365+notion, monitoring/service_monitor/env_analyzer through shared pool (#12979)

autobot-backend/api/marketplace_sources.py

@@ -199,6 +199,16 @@ async def _fetch_catalog_document(url: str) -> CatalogDocument:
     via `_resolve_safe_ip` (blocks RFC1918/loopback/link-local) and
     ``allow_redirects=False`` (defeats redirect-rebind). Raises
     HTTPException(400) on any fetch/parse/validation failure.
+
+    Issue #12979: deliberately NOT converted to the shared pooled client. The
+    ``_PinnedResolver``/``connector=`` below pin this session to the
+    pre-resolved public IP so the fetch cannot be DNS-rebound to a private
+    address between validation and connect. A connector is session-level —
+    the pooled ``HTTPClientManager`` owns exactly one shared ``TCPConnector``
+    and ``session.request()`` accepts no per-request override — so routing
+    this through the pool would silently drop the pin and reopen the
+    DNS-rebinding hole this function exists to close. Same carve-out category
+    as ``knowledge/connectors/oauth_flow.py::_post_token()``.
     """
     _validate_url_scheme(url)
     parsed = urlparse(url)

autobot-backend/api/monitoring.py

@@ -95,6 +95,13 @@ async def _query_prometheus_instant(query: str) -> float | None:
 
     Uses the singleton HTTP client for connection reuse (Issue #65).
 
+    Issue #12979: switched from ``await http_client.get(...)`` under a plain
+    ``async with`` to ``tracked_request()`` — the old form releases the
+    *connection* but leaves ``_active_requests`` permanently incremented
+    (the #12981 defect), since ``get()``/``request()`` delegate the
+    success-path decrement to the caller and this response is never
+    passed to ``decrement_active()``.
+
     Args:
         query: PromQL query string.
 
@@ -104,7 +111,8 @@ async def _query_prometheus_instant(query: str) -> float | None:
     try:
         http_client = get_http_client()
         params = {"query": query}
-        async with await http_client.get(
+        async with http_client.tracked_request(
+            "GET",
             f"{_PROMETHEUS_URL}/api/v1/query",
             params=params,
             timeout=aiohttp.ClientTimeout(total=10),
@@ -134,6 +142,10 @@ async def _query_prometheus_range(
 
     Uses the singleton HTTP client for connection reuse (Issue #65).
 
+    Issue #12979: switched from ``await http_client.get(...)`` under a plain
+    ``async with`` to ``tracked_request()`` — see ``_query_prometheus_instant``
+    for why (the #12981 counter-leak defect).
+
     Args:
         query: PromQL query string.
         start: Start of the query window.
@@ -151,7 +163,8 @@ async def _query_prometheus_range(
             "end": to_rfc3339(end),
             "step": step,
         }
-        async with await http_client.get(
+        async with http_client.tracked_request(
+            "GET",
             f"{_PROMETHEUS_URL}/api/v1/query_range",
             params=params,
             timeout=aiohttp.ClientTimeout(total=30),
@@ -224,6 +237,12 @@ async def _fetch_alertmanager_alerts() -> List[Dict[str, Any]]:
 
     Returns:
         List of active alerts in frontend-compatible format.
+
+    Issue #12979: routed through the shared pooled client's
+    ``tracked_request()`` instead of a per-request ``aiohttp.ClientSession``.
+    ``suppress_error_log=True`` since AlertManager being unreachable is an
+    expected, already-logged (WARNING) condition on this 10s-cached poll —
+    the manager's default ERROR log would just duplicate it.
     """
     async with _alertmanager_cache_lock:
         current_time = time.time()
@@ -233,19 +252,20 @@ async def _fetch_alertmanager_alerts() -> List[Dict[str, Any]]:
             return _alertmanager_cache["alerts"]
 
         try:
-            async with aiohttp.ClientSession() as session:
-                # AlertManager v2 API for active alerts
-                url = f"{ServiceURLs.ALERTMANAGER_API}/api/v2/alerts"
-                async with session.get(url, timeout=_ALERTMANAGER_TIMEOUT) as response:
-                    if response.status == 200:
-                        raw_alerts = await response.json()
-                        formatted_alerts = _format_alertmanager_alerts(raw_alerts)
-                        _alertmanager_cache["alerts"] = formatted_alerts
-                        _alertmanager_cache["timestamp"] = current_time
-                        return formatted_alerts
-                    else:
-                        logger.warning("AlertManager returned status %d", response.status)
-                        return _alertmanager_cache["alerts"]  # Return stale cache
+            # AlertManager v2 API for active alerts
+            url = f"{ServiceURLs.ALERTMANAGER_API}/api/v2/alerts"
+            async with get_http_client().tracked_request(
+                "GET", url, timeout=_ALERTMANAGER_TIMEOUT, suppress_error_log=True
+            ) as response:
+                if response.status == 200:
+                    raw_alerts = await response.json()
+                    formatted_alerts = _format_alertmanager_alerts(raw_alerts)
+                    _alertmanager_cache["alerts"] = formatted_alerts
+                    _alertmanager_cache["timestamp"] = current_time
+                    return formatted_alerts
+                else:
+                    logger.warning("AlertManager returned status %d", response.status)
+                    return _alertmanager_cache["alerts"]  # Return stale cache
         except asyncio.TimeoutError:
             logger.warning("AlertManager request timed out")
             return _alertmanager_cache["alerts"]

autobot-backend/api/service_monitor.py

@@ -22,6 +22,7 @@
     ServiceMonitorVMsResponse,
 )
 from autobot_shared.error_boundaries import ErrorCategory, with_error_handling
+from autobot_shared.http_client import get_http_client
 from autobot_shared.logging_manager import get_logger
 from autobot_shared.ssot_config import config as _ssot
 
@@ -38,13 +39,19 @@ async def _check_http_health(url: str) -> Tuple[str, str]:
 
     Returns ("online", "Healthy") on 2xx, ("offline", reason) otherwise.
     Issue #925: helper for get_service_statuses / get_vm_statuses.
+
+    Issue #12979: routed through the shared pooled client's
+    ``tracked_request()``. ``suppress_error_log=True`` — a service being
+    offline is an expected, already-handled condition here (health polling),
+    not an unexpected outbound-call failure.
     """
     try:
-        async with aiohttp.ClientSession(timeout=_HEALTH_TIMEOUT) as session:
-            async with session.get(url, ssl=False) as resp:
-                if resp.status < 300:
-                    return "online", "Healthy"
-                return "offline", f"HTTP {resp.status}"
+        async with get_http_client().tracked_request(
+            "GET", url, timeout=_HEALTH_TIMEOUT, ssl=False, suppress_error_log=True
+        ) as resp:
+            if resp.status < 300:
+                return "online", "Healthy"
+            return "offline", f"HTTP {resp.status}"
     except asyncio.TimeoutError:
         return "offline", "Timeout"
     except aiohttp.ClientConnectorError:

autobot-backend/code_analysis/src/env_analyzer.py

@@ -1343,18 +1343,26 @@ async def _check_ollama_health(self, host: str, port: str) -> bool:
         """Check if Ollama service is available.
 
         Helper for llm_filter_hardcoded (#633).
+
+        Issue #12979: routed through the shared pooled client's
+        ``tracked_request()``. ``suppress_error_log=True`` — Ollama being
+        unavailable is an expected condition this health probe already
+        handles by returning unfiltered results (logged at WARNING here).
         """
         import aiohttp
 
+        from autobot_shared.http_client import get_http_client
+
         try:
-            async with aiohttp.ClientSession() as session:
-                async with session.get(
-                    f"http://{host}:{port}/api/tags",
-                    timeout=aiohttp.ClientTimeout(total=5),
-                ) as resp:
-                    if resp.status != 200:
-                        logger.warning("Ollama not available, " "returning unfiltered results")
-                        return False
+            async with get_http_client().tracked_request(
+                "GET",
+                f"http://{host}:{port}/api/tags",
+                timeout=aiohttp.ClientTimeout(total=5),
+                suppress_error_log=True,
+            ) as resp:
+                if resp.status != 200:
+                    logger.warning("Ollama not available, " "returning unfiltered results")
+                    return False
         except Exception as e:
             logger.warning(f"Ollama health check failed: {e}, " "returning unfiltered results")
             return False
@@ -1381,7 +1389,7 @@ async def _process_llm_batches(
             prompt = self._build_llm_filter_prompt(batch)
 
             try:
-                true_indices = await self._call_ollama_filter(ollama_url, model, prompt, session=None)
+                true_indices = await self._call_ollama_filter(ollama_url, model, prompt)
                 for idx in true_indices:
                     if 0 <= idx < len(batch):
                         filtered_results.append(batch[idx])
@@ -1462,24 +1470,32 @@ async def _call_ollama_filter(
         url: str,
         model: str,
         prompt: str,
-        session: Any | None = None,
     ) -> list[int]:
         """
         Call Ollama API for filtering and parse response.
 
         Issue #633: Handles Ollama generate API response.
 
+        Issue #12979: routed through the shared pooled client's
+        ``tracked_request()`` instead of a per-call ``aiohttp.ClientSession``.
+        Dropped the ``session`` parameter — its "caller supplies its own
+        session" branch had exactly one caller (``_process_llm_batches``,
+        same file) and it always passed ``session=None``, so the reused-
+        session path was dead code; the pooled client now provides the
+        connection reuse that parameter existed to allow.
+
         Args:
             url: Ollama generate endpoint URL
             model: Model name to use
             prompt: The filter prompt
-            session: Optional aiohttp session (creates new if None)
 
         Returns:
             List of 1-indexed line numbers that are true issues
         """
         import aiohttp
 
+        from autobot_shared.http_client import get_http_client
+
         payload = {
             "model": model,
             "prompt": prompt,
@@ -1490,24 +1506,17 @@ async def _call_ollama_filter(
             },
         }
 
-        should_close = session is None
-        if session is None:
-            session = aiohttp.ClientSession()
-
-        try:
-            async with session.post(url, json=payload, timeout=aiohttp.ClientTimeout(total=60)) as response:
-                if response.status != 200:
-                    error_text = await response.text()
-                    raise Exception(f"Ollama API error {response.status}: {error_text}")
-
-                result = await response.json()
-                response_text = result.get("response", "").strip()
+        async with get_http_client().tracked_request(
+            "POST", url, json=payload, timeout=aiohttp.ClientTimeout(total=60)
+        ) as response:
+            if response.status != 200:
+                error_text = await response.text()
+                raise Exception(f"Ollama API error {response.status}: {error_text}")
 
-                return self._parse_llm_filter_response(response_text)
+            result = await response.json()
+            response_text = result.get("response", "").strip()
 
-        finally:
-            if should_close:
-                await session.close()
+            return self._parse_llm_filter_response(response_text)
 
     def _parse_llm_filter_response(self, response_text: str) -> list[int]:
         """

autobot-backend/integrations/base.py

@@ -120,23 +120,32 @@ async def _make_request(
         """Make an HTTP request to the external service.
 
         Helper for test_connection and execute_action.
+
+        Issue #12979: routed through the shared pooled client instead of a
+        per-request ``aiohttp.ClientSession``. Never raises on HTTP status —
+        callers (including ``whatsapp_integration.py``) inspect the returned
+        ``status_code`` themselves — so ``tracked_request()`` is used rather
+        than ``get_json()``/``post_json()``, which would raise on non-2xx.
         """
         import aiohttp
 
+        from autobot_shared.http_client import get_http_client
+
         merged_headers = headers or {}
         if self.config.api_key:
             merged_headers.setdefault("Authorization", f"Bearer {self.config.api_key}")
 
         try:
             timeout_obj = aiohttp.ClientTimeout(total=timeout)
-            async with aiohttp.ClientSession(timeout=timeout_obj) as session:
-                async with session.request(method, url, headers=merged_headers, json=json_data) as resp:
-                    body = await resp.json()
-                    return {
-                        "status_code": resp.status,
-                        "body": body,
-                        "headers": dict(resp.headers),
-                    }
+            async with get_http_client().tracked_request(
+                method, url, headers=merged_headers, json=json_data, timeout=timeout_obj
+            ) as resp:
+                body = await resp.json()
+                return {
+                    "status_code": resp.status,
+                    "body": body,
+                    "headers": dict(resp.headers),
+                }
         except aiohttp.ClientError as exc:
             self.logger.warning("Request to %s failed: %s", url, exc)
             return {"status_code": 0, "body": {}, "error": str(exc)}