feat: Add UV support for Ray Client mode

This commit enables UV package manager integration for Ray Client
connections, fixing issue #57991.

Previously, the RAY_RUNTIME_ENV_HOOK was disabled for Ray Client mode,
preventing UV from detecting and installing dependencies from
pyproject.toml and uv.lock files. This caused ModuleNotFoundError
when using Ray Client with UV environments.

Changes:
- Add _apply_uv_hook_for_client() to detect UV on client side
- Modify ClientContext.connect() to apply UV hook before connection
- UV hook now runs client-side where 'uv run' process exists
- Server-side hooks remain disabled for safety

The fix detects 'uv run' on the client machine and propagates the
UV configuration (py_executable, working_dir) to cluster workers,
allowing them to install dependencies via UV.

Tests added:
- test_ray_client_uv_detection: Verify UV detection and propagation
- test_ray_client_uv_no_detection_without_uv: Verify normal operation
- test_ray_client_uv_with_job_config: Test with JobConfig
- test_uv_run_ray_client_mode: End-to-end UV installation test

Fixes #57991

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: preneond

python/ray/tests/test_client.py

@@ -985,5 +985,114 @@ def test_internal_kv_in_proxy_mode(call_ray_start_shared):
     assert client_api._internal_kv_get(b"key") is None
 
 
+def test_ray_client_uv_hook_detection():
+    """Test that _apply_uv_hook_for_client correctly detects and applies UV config.
+
+    Related to: https://github.com/ray-project/ray/issues/57991
+    """
+    from unittest.mock import patch
+
+    from ray.util.client import _apply_uv_hook_for_client
+
+    # Test 1: UV detected - should set py_executable
+    with patch(
+        "ray._private.runtime_env.uv_runtime_env_hook._get_uv_run_cmdline"
+    ) as mock_uv:
+        mock_uv.return_value = [
+            "uv",
+            "run",
+            "--locked",
+            "--python",
+            "3.11",
+            "script.py",
+        ]
+
+        result = _apply_uv_hook_for_client({"working_dir": "/tmp/test"})
+
+        assert "py_executable" in result, "py_executable should be set when UV detected"
+        assert "uv run" in result["py_executable"], "should contain 'uv run'"
+        assert "--locked" in result["py_executable"], "should preserve --locked flag"
+        # working_dir should be preserved (not overwritten)
+        assert result["working_dir"] == "/tmp/test"
+
+    # Test 2: No UV detected - should return runtime_env unchanged
+    with patch(
+        "ray._private.runtime_env.uv_runtime_env_hook._get_uv_run_cmdline"
+    ) as mock_uv:
+        mock_uv.return_value = None
+
+        original = {"working_dir": "/tmp/test"}
+        result = _apply_uv_hook_for_client(original)
+
+        assert "py_executable" not in result, "py_executable should not be set"
+        assert result == original
+
+    # Test 3: None runtime_env - should handle gracefully
+    with patch(
+        "ray._private.runtime_env.uv_runtime_env_hook._get_uv_run_cmdline"
+    ) as mock_uv:
+        mock_uv.return_value = None
+
+        result = _apply_uv_hook_for_client(None)
+        assert result is None
+
+    # Test 4: Hook failure - should return original and not crash
+    with patch(
+        "ray._private.runtime_env.uv_runtime_env_hook._get_uv_run_cmdline",
+        side_effect=RuntimeError("mock error"),
+    ):
+        original = {"working_dir": "/tmp/test"}
+        result = _apply_uv_hook_for_client(original)
+        assert result == original, "should return original on error"
+
+
+def test_ray_client_uv_no_detection_without_uv(call_ray_start_shared):
+    """Test that Ray Client works normally when UV is not detected."""
+    # Ensure clean state (previous tests may leave Ray initialized)
+    import ray as ray_module
+
+    ray_module.shutdown()
+    with ray_start_client_server_for_address(call_ray_start_shared) as ray:
+        # Connect without UV (normal case)
+
+        @ray.remote
+        def simple_task():
+            return "success"
+
+        # Should work normally
+        result = ray.get(simple_task.remote())
+        assert result == "success"
+
+
+def test_ray_client_uv_hook_with_existing_runtime_env():
+    """Test UV hook correctly merges with existing runtime_env settings.
+
+    Related to: https://github.com/ray-project/ray/issues/57991
+    """
+    from unittest.mock import patch
+
+    from ray.util.client import _apply_uv_hook_for_client
+
+    with patch(
+        "ray._private.runtime_env.uv_runtime_env_hook._get_uv_run_cmdline"
+    ) as mock_uv:
+        mock_uv.return_value = ["uv", "run", "script.py"]
+
+        # Existing runtime_env with working_dir should be preserved
+        runtime_env = {
+            "working_dir": "/my/custom/dir",
+            "env_vars": {"MY_VAR": "value"},
+        }
+        result = _apply_uv_hook_for_client(runtime_env)
+
+        assert "py_executable" in result
+        assert (
+            result["working_dir"] == "/my/custom/dir"
+        ), "User's working_dir should take precedence"
+        assert (
+            result["env_vars"]["MY_VAR"] == "value"
+        ), "Existing env_vars should be preserved"
+
+
 if __name__ == "__main__":
     sys.exit(pytest.main(["-sv", __file__]))

python/ray/tests/test_runtime_env_uv_run.py

@@ -499,5 +499,65 @@ def f():
         assert json.load(f) == {"working_dir_files": os.listdir(working_dir)}
 
 
+@pytest.mark.skipif(sys.platform == "win32", reason="Not ported to Windows yet.")
+def test_uv_run_ray_client_mode(shutdown_only, tmp_working_dir):
+    """Test that UV environment works with Ray Client mode.
+
+    This verifies the fix for: https://github.com/ray-project/ray/issues/57991
+    """
+    from unittest.mock import patch
+
+    tmp_dir = tmp_working_dir
+
+    # Start Ray cluster
+    ray.init(num_cpus=1)
+
+    # Start Ray Client server
+    import ray.util.client.server.server as ray_client_server
+
+    server_handle = ray_client_server.serve("localhost", 50052)
+
+    try:
+        # Simulate running under 'uv run' by mocking the process tree
+        with patch(
+            "ray._private.runtime_env.uv_runtime_env_hook._get_uv_run_cmdline"
+        ) as mock_uv:
+            mock_uv.return_value = [
+                "uv",
+                "run",
+                "--python-preference=only-system",
+                "script.py",
+            ]
+
+            # Connect via Ray Client
+            import ray as ray_client
+
+            ray_client.util.connect(
+                "localhost:50052",
+                runtime_env={
+                    "working_dir": tmp_dir,
+                    # Use system environment to find Ray installation
+                    "env_vars": {"PYTHONPATH": ":".join(sys.path)},
+                },
+            )
+
+            @ray_client.remote
+            def emojize():
+                import emoji
+
+                return emoji.emojize("Ray rocks :thumbs_up:")
+
+            # This should work because UV hook detected and propagated UV config
+            result = ray_client.get(emojize.remote())
+            assert (
+                result == "Ray rocks 👍"
+            ), "UV should have installed emoji package on workers"
+
+            ray_client.util.disconnect()
+    finally:
+        server_handle.stop(0)
+        ray.shutdown()
+
+
 if __name__ == "__main__":
     sys.exit(pytest.main(["-sv", __file__]))

python/ray/util/client/__init__.py

@@ -16,6 +16,45 @@
 logger = logging.getLogger(__name__)
 
 
+def _apply_uv_hook_for_client(runtime_env: Optional[Dict[str, Any]]) -> Dict[str, Any]:
+    """Apply UV runtime env hook on client side before connection.
+
+    This allows Ray Client to support 'uv run' environments by detecting
+    the UV configuration on the client side and propagating it to workers.
+    The hook only runs if RAY_ENABLE_UV_RUN_RUNTIME_ENV is enabled and
+    the client is running under 'uv run'.
+
+    Args:
+        runtime_env: The runtime environment dict to potentially modify.
+
+    Returns:
+        Modified runtime_env dict with UV configuration if detected,
+        otherwise the original runtime_env unchanged.
+    """
+    try:
+        if not ray_constants.RAY_ENABLE_UV_RUN_RUNTIME_ENV:
+            return runtime_env
+
+        from ray._private.runtime_env.uv_runtime_env_hook import (
+            _get_uv_run_cmdline,
+            hook,
+        )
+
+        cmdline = _get_uv_run_cmdline()
+        if cmdline:
+            # UV environment detected on client side
+            logger.debug(f"UV environment detected: {cmdline}")
+            return hook(runtime_env)
+    except Exception as e:
+        # Log warning but don't fail connection
+        logger.warning(
+            f"Failed to apply UV runtime env hook for Ray Client: {e}. "
+            "UV environment will not be propagated to workers."
+        )
+
+    return runtime_env
+
+
 class _ClientContext:
     def __init__(self):
         from ray.util.client.api import _ClientAPI
@@ -74,7 +113,24 @@ def connect(
         if ray_init_kwargs is None:
             ray_init_kwargs = {}
 
-        # NOTE(architkulkarni): env_hook is not supported with Ray Client.
+        # Apply UV hook client-side before connection (if UV detected).
+        # This allows UV environments to work with Ray Client by detecting
+        # the UV configuration on the client machine and propagating it to
+        # the cluster workers. See: https://github.com/ray-project/ray/issues/57991
+        runtime_env = ray_init_kwargs.get("runtime_env")
+        if job_config and job_config.runtime_env:
+            # Prefer job_config runtime_env if provided
+            runtime_env = runtime_env or job_config.runtime_env
+
+        runtime_env = _apply_uv_hook_for_client(runtime_env)
+
+        if runtime_env:
+            ray_init_kwargs["runtime_env"] = runtime_env
+            if job_config:
+                job_config.set_runtime_env(runtime_env)
+
+        # NOTE(architkulkarni): Custom env_hook is not supported with Ray Client.
+        # However, UV hook is now applied client-side above.
         ray_init_kwargs["_skip_env_hook"] = True
 
         if ray_init_kwargs.get("logging_level") is not None: