Fix Ray Client when 'uv run' runtime environment is used (#51683)

<!-- Thank you for your contribution! Please review
https://github.com/ray-project/ray/blob/master/CONTRIBUTING.rst before
opening a pull request. -->

<!-- Please add a reviewer to the assignee section when you create a PR.
If you don't have the access to it, we will shortly find a reviewer and
assign them to your PR. -->

## Why are these changes needed?

Fixes #51368

The problem was happening because we check the command line of the Ray
Client server, but when used with `py_executable` equal to `uv run`, the
command line would be `uv run -m ray.util.client.server ...` and not
`python -m ray.util.client.server`, so the check needed to be adapted.

## Related issue number

<!-- For example: "Closes #1234" -->

## Checks

- [ ] I've signed off every commit(by using the -s flag, i.e., `git
commit -s`) in this PR.
- [ ] I've run `scripts/format.sh` to lint the changes in this PR.
- [ ] I've included any doc changes needed for
https://docs.ray.io/en/master/.
- [ ] I've added any new APIs to the API Reference. For example, if I
added a
method in Tune, I've added it in `doc/source/tune/api/` under the
           corresponding `.rst` file.
- [ ] I've made sure the tests are passing. Note that there might be a
few flaky tests, see the recent failures at https://flakey-tests.ray.io/
- Testing Strategy
   - [ ] Unit tests
   - [ ] Release tests
   - [ ] This PR is not tested :(

Author: pcmoritz

python/ray/tests/test_client.py

@@ -29,6 +29,7 @@
 from ray.tests.conftest import call_ray_start_context
 from ray.util.client.common import OBJECT_TRANSFER_CHUNK_SIZE, ClientObjectRef
 from ray.util.client.ray_client_helpers import (
+    ray_start_client_server,
     ray_start_client_server_for_address,
 )
 
@@ -547,6 +548,21 @@ def f(x):
         assert ray.get(a.doit.remote()) == "foo"
 
 
+# Regression test for https://github.com/ray-project/ray/pull/51683
+def test_runtime_env_py_executable(ray_start_regular):
+    """Test that Ray Client works with a custom py_executable."""
+
+    with ray_start_client_server(
+        ray_init_kwargs={"runtime_env": {"py_executable": sys.executable + " -q"}}
+    ) as ray:
+
+        @ray.remote
+        def f():
+            return "hi"
+
+        assert ray.get(f.remote()) == "hi"
+
+
 def test_basic_named_actor(call_ray_start_shared):
     """Test that ray.get_actor() can create and return a detached actor."""
     with ray_start_client_server_for_address(call_ray_start_shared) as ray:

python/ray/tests/test_client_proxy.py

@@ -321,9 +321,8 @@ def modify_namespace(job_config: JobConfig):
         (["ipython", "-m", "ray.util.client.server"], True),
         (["ipython -m ray.util.client.server"], True),
         (["ipython -m", "ray.util.client.server"], True),
-        (["bash", "ipython", "-m", "ray.util.client.server"], False),
-        (["bash", "ipython -m ray.util.client.server"], False),
-        (["python", "-m", "bash", "ipython -m ray.util.client.server"], False),
+        (["bash", "-c", "ipython -m ray.util.client.server"], True),
+        (["python", "-m", "bash", "ipython"], False),
     ],
 )
 def test_match_running_client_server(test_case):

python/ray/util/client/server/proxier.py

@@ -100,14 +100,11 @@ def set_result(self, proc: Optional[ProcessInfo]) -> None:
 def _match_running_client_server(command: List[str]) -> bool:
     """
     Detects if the main process in the given command is the RayClient Server.
-    This works by ensuring that the the first three arguments are similar to:
-        <python> -m ray.util.client.server
+    This works by ensuring that the command is of the form:
+        <py_executable> -m ray.util.client.server <args>
     """
     flattened = " ".join(command)
-    rejoined = flattened.split()
-    if len(rejoined) < 3:
-        return False
-    return rejoined[1:3] == ["-m", "ray.util.client.server"]
+    return "-m ray.util.client.server" in flattened
 
 
 class ProxyManager: