undo

Author: xzrderek

eval_protocol/adapters/fireworks_tracing.py

@@ -264,7 +264,6 @@ def __init__(
         self.project_id = project_id
         self.base_url = base_url.rstrip("/")
         self.timeout = timeout
-        self._session = requests.Session()
 
     def search_logs(self, tags: List[str], limit: int = 100, hours_back: int = 24) -> List[Dict[str, Any]]:
         """Fetch logs from Fireworks tracing gateway /logs endpoint.
@@ -288,14 +287,14 @@ def search_logs(self, tags: List[str], limit: int = 100, hours_back: int = 24) -
         last_error: Optional[str] = None
         for url in urls_to_try:
             try:
-                with self._session.get(url, params=params, timeout=self.timeout, headers=headers) as response:
-                    if response.status_code == 404:
-                        # Try next variant (must close response to release connection)
-                        last_error = f"404 for {url}"
-                        continue
-                    response.raise_for_status()
-                    data = response.json() or {}
-                    break
+                response = requests.get(url, params=params, timeout=self.timeout, headers=headers)
+                if response.status_code == 404:
+                    # Try next variant
+                    last_error = f"404 for {url}"
+                    continue
+                response.raise_for_status()
+                data = response.json() or {}
+                break
             except requests.exceptions.RequestException as e:
                 last_error = str(e)
                 continue
@@ -413,20 +412,22 @@ def get_evaluation_rows(
 
         result = None
         try:
-            with self._session.get(url, params=params, timeout=self.timeout, headers=headers) as response:
-                if response.status_code >= 400:
-                    error_msg: str = response.text
-                    try:
-                        payload = response.json()
-                        if isinstance(payload, dict) and "detail" in payload:
-                            detail = payload.get("detail")
-                            if detail:
-                                error_msg = str(detail)
-                    except Exception:
-                        pass
-                    logger.error("Failed to fetch traces from proxy (HTTP %s): %s", response.status_code, error_msg)
-                    return eval_rows
-                result = response.json()
+            response = requests.get(url, params=params, timeout=self.timeout, headers=headers)
+            response.raise_for_status()
+            result = response.json()
+        except requests.exceptions.HTTPError as e:
+            error_msg = str(e)
+
+            # Try to extract detail message from response
+            if e.response is not None:
+                try:
+                    error_detail = e.response.json().get("detail", {})
+                    error_msg = error_detail or e.response.text
+                except Exception:  # In case e.response.json() fails
+                    error_msg = f"Proxy error: {e.response.text}"
+
+            logger.error("Failed to fetch traces from proxy (HTTP %s): %s", e.response.status_code, error_msg)
+            return eval_rows
         except requests.exceptions.RequestException as e:
             # Non-HTTP errors (network issues, timeouts, etc.)
             logger.error("Failed to fetch traces from proxy: %s", str(e))
@@ -450,10 +451,3 @@ def get_evaluation_rows(
 
         logger.info("Successfully converted %d traces to evaluation rows", len(eval_rows))
         return eval_rows
-
-    def close(self) -> None:
-        """Close underlying HTTP resources."""
-        try:
-            self._session.close()
-        except Exception:
-            pass

eval_protocol/pytest/remote_rollout_processor.py

@@ -54,7 +54,6 @@ def __init__(
         self._timeout_seconds = timeout_seconds
         self._output_data_loader = output_data_loader or default_fireworks_output_data_loader
         self._tracing_adapter = FireworksTracingAdapter(base_url=self._model_base_url)
-        self._session = requests.Session()
 
     def __call__(self, rows: List[EvaluationRow], config: RolloutProcessorConfig) -> List[asyncio.Task[EvaluationRow]]:
         tasks: List[asyncio.Task[EvaluationRow]] = []
@@ -95,8 +94,8 @@ async def _process_row(row: EvaluationRow) -> EvaluationRow:
             def _post_init() -> None:
                 url = f"{remote_base_url}/init"
                 try:
-                    with self._session.post(url, json=init_payload.model_dump(), timeout=300) as r:
-                        r.raise_for_status()
+                    r = requests.post(url, json=init_payload.model_dump(), timeout=300)
+                    r.raise_for_status()
                 except requests.exceptions.Timeout:
                     raise TimeoutError(
                         f"The /init endpoint tried {url} with {init_payload.model_dump()} but timed out after 300 seconds."
@@ -109,9 +108,9 @@ def _post_init() -> None:
 
             def _get_status() -> Dict[str, Any]:
                 url = f"{remote_base_url}/status"
-                with self._session.get(url, params={"rollout_id": row.execution_metadata.rollout_id}, timeout=15) as r:
-                    r.raise_for_status()
-                    return r.json()
+                r = requests.get(url, params={"rollout_id": row.execution_metadata.rollout_id}, timeout=15)
+                r.raise_for_status()
+                return r.json()
 
             continue_polling_status = True
             while time.time() < deadline:
@@ -205,12 +204,4 @@ async def _sem_wrapper(r: EvaluationRow) -> EvaluationRow:
         return tasks
 
     def cleanup(self) -> None:
-        try:
-            self._tracing_adapter.close()
-        except Exception:
-            pass
-        try:
-            self._session.close()
-        except Exception:
-            pass
         return None