debug

Author: xzrderek

eval_protocol/mcp/client/connection.py

@@ -88,12 +88,6 @@ async def _prewarm_tools_cache(self, session: MCPSession) -> None:
         """
         cache_key = session.base_url
 
-        # Fast path: if cache already exists, return immediately (no lock)
-        if cache_key in self._tools_cache:
-            logger.debug(f"Tools cache already exists for {cache_key}")
-            return
-
-        # Slow path: need to create cache (use lock only for creation)
         async with self._tools_cache_lock:
             # Only fetch tools if not already cached for this base_url
             if cache_key not in self._tools_cache:
@@ -123,7 +117,7 @@ async def reset_session(self, session: MCPSession) -> None:
         headers = {"mcp-session-id": session.session_id}
         body = {"seed": session.seed}
 
-        timeout = httpx.Timeout(3.0)
+        timeout = httpx.Timeout(15.0)
         async with httpx.AsyncClient(timeout=timeout) as client:
             resp = await client.post(url, headers=headers, json=body)
             resp.raise_for_status()
@@ -145,23 +139,16 @@ async def discover_tools(self, session: MCPSession) -> List[Dict]:
 
         cache_key = session.base_url
 
-        # Fast path: Check cache first without lock (safe for reads)
-        if cache_key in self._tools_cache:
-            cached_tools = self._tools_cache[cache_key]
-            logger.debug(f"Using cached tools for session {session.session_id} ({len(cached_tools)} tools)")
-            return cached_tools
-
-        # Slow path: Cache miss - use lock only for writing
+        # Check cache first (should be pre-warmed during initialization)
         async with self._tools_cache_lock:
-            # Double-check pattern: another task might have cached it while we waited
             if cache_key in self._tools_cache:
                 cached_tools = self._tools_cache[cache_key]
                 logger.debug(f"Using cached tools for session {session.session_id} ({len(cached_tools)} tools)")
                 return cached_tools
 
-            # Fallback: if cache miss (shouldn't happen with pre-warming), fetch directly
-            logger.warning(f"Cache miss for {cache_key} - this shouldn't happen with pre-warming")
-            mcp_session = session._mcp_session
+        # Fallback: if cache miss (shouldn't happen with pre-warming), fetch directly
+        logger.warning(f"Cache miss for {cache_key} - this shouldn't happen with pre-warming")
+        mcp_session = session._mcp_session
 
         tools_response = await mcp_session.list_tools()
         tools = tools_response.tools if hasattr(tools_response, "tools") else []
@@ -213,7 +200,6 @@ async def get_initial_state(self, session: MCPSession) -> Any:
         logger.info(f"### 🌟 GET_INITIAL_STATE_START: timestamp: {method_start}, session_id: {session_id_short}...")
 
         if not session._mcp_session:
-            logger.error(f"### ❌ SESSION_NOT_INITIALIZED: session_id: {session_id_short}")
             raise RuntimeError("Session not initialized")
 
         # Try to get initial state from control plane endpoint first

eval_protocol/mcp/execution/manager.py

@@ -288,13 +288,13 @@ async def _execute_rollout(
                         )
                         user_content = user_message.content if user_message.content else ""
 
-                    user_prompt = envs.format_user_prompt(rollout_idx, user_content)
-                    conversation_history.append({"role": "user", "content": user_prompt})
+                        user_prompt = envs.format_user_prompt(rollout_idx, user_content)
+                        conversation_history.append({"role": "user", "content": user_prompt})
 
-                    # Check if user simulator signaled termination
-                    if UserSimulator.is_stop(user_message):
-                        trajectory.terminated = True
-                        trajectory.termination_reason = TerminationReason.USER_STOP
+                        # Check if user simulator signaled termination
+                        if UserSimulator.is_stop(user_message):
+                            trajectory.terminated = True
+                            trajectory.termination_reason = TerminationReason.USER_STOP
 
                 # In each turn: keep looping until assistant is ready to provide final response
                 while not turn_completed and not trajectory.terminated:

eval_protocol/mcp/mcpgym.py

@@ -28,9 +28,7 @@
 from typing import Any, Callable, Dict, Optional, Tuple
 
 import uvicorn
-
-# from mcp.server.fastmcp import Context, FastMCP
-from fastmcp import Context, FastMCP
+from mcp.server.fastmcp import Context, FastMCP
 from pydantic import BaseModel
 from starlette.requests import Request
 from starlette.responses import JSONResponse
@@ -104,8 +102,11 @@ def __init__(
         self.adapter = adapter
 
         # Create FastMCP server
-        self.mcp = FastMCP(name=server_name)
-
+        self.mcp = FastMCP(
+            server_name,
+            host="0.0.0.0",
+            port=int(os.environ.get("PORT", 8000)),
+        )
         # Store host and port for later use in run() method
         self.host = "0.0.0.0"
         self.port = int(os.environ.get("PORT", 8000))
@@ -129,6 +130,7 @@ def __init__(
 
         self.pool = ThreadPoolExecutor(max_workers=max_workers)
 
+        # Reset with seed if provided
         self.env, self.obs, _info = self._new_env(seed=seed)
 
         # Register tools and control plane endpoints
@@ -220,7 +222,8 @@ async def reset_session_endpoint(request: Request) -> JSONResponse:
             if not session_id:
                 return JSONResponse({"error": "Missing mcp-session-id header"}, status_code=400)
             if session_id in self.sessions:
-                env, obs, _ = self._new_env(seed=seed)
+                loop = asyncio.get_running_loop()
+                env, obs, info = await loop.run_in_executor(self.pool, self._new_env, seed)
                 with self.session_lock:
                     self.sessions[session_id] = {
                         "env": env,
@@ -269,17 +272,10 @@ async def endpoint_handler(request: Request) -> JSONResponse:
                                     {"error": f"Session {session_id} not found"},
                                     status_code=404,
                                 )
-                            start_time = time.time()
-                            logger.info(
-                                f"### 🔍 NEW_ENV_START: timestamp: {start_time}, session_id: {session_id[:8] if len(session_id) > 8 else session_id}..."
-                            )
+
                             loop = asyncio.get_running_loop()
                             env, obs, info = await loop.run_in_executor(self.pool, self._new_env, None)
-                            # env, obs, info = self._new_env(None)
-                            end_time = time.time()
-                            logger.info(
-                                f"### 🔍 NEW_ENV_END: timestamp: {end_time}, elapsed: {end_time - start_time:.6f}s, session_id: {session_id[:8] if len(session_id) > 8 else session_id}..."
-                            )
+
                             # Initialize session state with extracted seed from session ID
                             session_data = {
                                 "env": env,
@@ -294,6 +290,7 @@ async def endpoint_handler(request: Request) -> JSONResponse:
                             result = await func(session_data=session_data)
                         else:
                             result = func(session_data=session_data)
+
                         return JSONResponse(result)
 
                     except Exception as e:
@@ -484,78 +481,26 @@ def get_info_endpoint(self, session_data: Dict[str, Any]) -> Dict[str, Any]:
     @control_plane_endpoint("/control/initial_state")
     async def get_initial_state_endpoint(self, session_data: Dict[str, Any]) -> Dict[str, Any]:
         """Get initial state for this session."""
-        endpoint_start = time.time()
         session_id = session_data.get("session_id", "unknown")
-        logger.info(
-            f"### 🌟 ENDPOINT_START: get_initial_state_endpoint, timestamp: {endpoint_start}, session_id: {session_id[:8] if len(session_id) > 8 else session_id}..."
-        )
-
-        env_check_start = time.time()
-        logger.info(
-            f"### 🔍 ENV_CHECK_START: timestamp: {env_check_start}, elapsed: {env_check_start - endpoint_start:.6f}s, session_id: {session_id[:8] if len(session_id) > 8 else session_id}..."
-        )
-
         env = session_data.get("env")
         obs = session_data.get("obs")
-
-        env_check_end = time.time()
-        logger.info(
-            f"### 🔍 ENV_CHECK_END: timestamp: {env_check_end}, elapsed: {env_check_end - endpoint_start:.6f}s, duration: {env_check_end - env_check_start:.6f}s, env: {env is not None}, obs: {obs is not None}, session_id: {session_id[:8] if len(session_id) > 8 else session_id}..."
-        )
-
         if env and obs is not None:
-            format_start = time.time()
-            logger.info(
-                f"### 🔄 FORMAT_OBS_START: timestamp: {format_start}, elapsed: {format_start - endpoint_start:.6f}s, session_id: {session_id[:8] if len(session_id) > 8 else session_id}..."
-            )
-
             try:
                 formatted_obs = self.format_observation(obs, env)
-
-                format_end = time.time()
-                logger.info(
-                    f"### 🔄 FORMAT_OBS_END: timestamp: {format_end}, elapsed: {format_end - endpoint_start:.6f}s, duration: {format_end - format_start:.6f}s, session_id: {session_id[:8] if len(session_id) > 8 else session_id}..."
-                )
-
-                endpoint_end = time.time()
-                logger.info(
-                    f"### ✅ ENDPOINT_SUCCESS_END: timestamp: {endpoint_end}, total_duration: {endpoint_end - endpoint_start:.6f}s, session_id: {session_id[:8] if len(session_id) > 8 else session_id}..."
-                )
-
                 return formatted_obs
             except Exception as e:
-                error_time = time.time()
-                logger.error(
-                    f"### ❌ FORMAT_OBS_ERROR: timestamp: {error_time}, elapsed: {error_time - endpoint_start:.6f}s, error: {str(e)}, session_id: {session_id[:8] if len(session_id) > 8 else session_id}..."
-                )
-
+                logger.error(f"❌ Error in format_observation: {e}")
                 return {
                     "error": f"Failed to format observation: {str(e)}",
                     "observation_type": str(type(obs)),
                     "session_id": session_data.get("session_id", "unknown"),
                 }
         else:
-            fallback_start = time.time()
-            logger.info(
-                f"### 🔄 FALLBACK_START: timestamp: {fallback_start}, elapsed: {fallback_start - endpoint_start:.6f}s, session_id: {session_id[:8] if len(session_id) > 8 else session_id}..."
-            )
-
             # Fallback if session data is not available
             result = {
                 "observation": "session_not_initialized",
                 "session_id": session_data.get("session_id", "unknown"),
             }
-
-            fallback_end = time.time()
-            logger.info(
-                f"### 🔄 FALLBACK_END: timestamp: {fallback_end}, elapsed: {fallback_end - endpoint_start:.6f}s, duration: {fallback_end - fallback_start:.6f}s, session_id: {session_id[:8] if len(session_id) > 8 else session_id}..."
-            )
-
-            endpoint_end = time.time()
-            logger.info(
-                f"### ✅ ENDPOINT_FALLBACK_END: timestamp: {endpoint_end}, total_duration: {endpoint_end - endpoint_start:.6f}s, session_id: {session_id[:8] if len(session_id) > 8 else session_id}..."
-            )
-
             return result
 
     def _get_session_control_plane_from_data(self, session_data: Dict[str, Any]) -> Dict[str, Any]:
@@ -623,9 +568,9 @@ async def run_with_high_concurrency():
 
                 config = uvicorn.Config(
                     starlette_app,
-                    host=self.host,
-                    port=self.port,
-                    log_level="info",  # Use default log level instead of accessing settings
+                    host=self.mcp.settings.host,
+                    port=self.mcp.settings.port,
+                    log_level=self.mcp.settings.log_level.lower(),  # Use default log level instead of accessing settings
                     proxy_headers=True,
                     forwarded_allow_ips="*",
                     # HIGH CONCURRENCY SETTINGS

eval_protocol/pytest/default_mcp_gym_rollout_processor.py

@@ -224,14 +224,11 @@ async def default_mcp_gym_rollout_processor(
         )
 
         # Create MCP environments directly from evaluation_rows
-        print("DEBUG1", time.time())
         envs = await ep.make(
             "http://localhost:9700/mcp/",
             evaluation_rows=rows,
             model_id=policy.model_id,
         )
-        print("DEBUG2", time.time())
-        print("max_concurrent_rollouts", config.max_concurrent_rollouts)
 
         # Run rollout with environments and policy
         evaluation_rows = await ep.rollout(

examples/frozen_lake_mcp/frozen_lake_mcp.py

@@ -19,8 +19,8 @@
 
 from typing import Any, Dict, Optional
 
-from fastmcp import Context
 from frozen_lake_adapter import FrozenLakeAdapter
+from mcp.server.fastmcp import Context
 
 from eval_protocol.mcp import McpGym
 from eval_protocol.mcp.mcpgym import control_plane_endpoint

examples/tau2_mcp/airplane_environment/airline_environment.py

@@ -37,13 +37,9 @@ def __init__(self, config: Optional[Dict[str, Any]] = None):
     def reset(self, seed: Optional[int] = None) -> Tuple[Dict[str, Any], Dict[str, Any]]:
         """Reset the environment to initial state"""
         logger.info("🔄 Resetting airline environment - reloading database from disk")
-        start_time = time.time()
         self.db = FlightDB.load(AIRLINE_DB_PATH)
         self.airline_tools = AirlineTools(self.db)
 
-        end_time = time.time()
-        logger.info(f"11RESET TOOK {end_time - start_time:.2f} seconds, called at {start_time}")
-
         return {}, {}
 
     def step(self, action: Dict[str, Any]) -> Tuple[Dict[str, Any], float, bool, bool, Dict[str, Any]]:

examples/tau2_mcp/tau2_mcp.py

@@ -12,9 +12,7 @@
 from typing import Annotated, Any, Dict, List, Optional
 
 from airplane_environment.airline_environment import AirlineEnvironment
-
-# from mcp.server.fastmcp import Context
-from fastmcp import Context
+from mcp.server.fastmcp import Context
 from mock_environment.mock_environment import MockEnvironment
 from pydantic import Field
 from retail_environment.retail_environment import RetailEnvironment