[core] symmetric-run fix (#58337)

## Description

This fixes the symmetric-run cli workflow.

Right now if you use `ray symmetric-run` on 2.51 like 
```
 ray symmetric-run --address 127.0.0.1:6379 -- python my_script.py   
```

it will throw since the `symmetric-run` arg is not caught. This was only
caught once it became part of the CLI.

## Related issues

## Additional information
> Optional: Add implementation details, API changes, usage examples,
screenshots, etc.

---------

Signed-off-by: Richard Liaw <[email protected]>

Author: richardliaw

python/ray/scripts/symmetric_run.py

@@ -133,6 +133,10 @@ def curate_and_validate_ray_start_args(run_and_start_args: List[str]) -> List[st
 @click.argument("ray_args_and_entrypoint", nargs=-1, type=click.UNPROCESSED)
 def symmetric_run(address, min_nodes, ray_args_and_entrypoint):
     all_args = sys.argv[1:]
+
+    if all_args and all_args[0] == "symmetric-run":
+        all_args = all_args[1:]
+
     try:
         separator = all_args.index("--")
     except ValueError:

python/ray/tests/test_symmetric_run.py

@@ -52,7 +52,7 @@ def test_symmetric_run_basic_interface(monkeypatch, cleanup_ray):
         with _setup_mock_network_utils("127.0.0.1", "127.0.0.1"):
             args = ["--address", "127.0.0.1:6379", "--", "echo", "test"]
 
-            with patch("sys.argv", ["ray.scripts.symmetric_run", *args]):
+            with patch("sys.argv", ["/bin/ray", "symmetric-run", *args]):
                 # Test basic symmetric_run call using CliRunner
                 result = runner.invoke(symmetric_run, args)
                 assert result.exit_code == 0
@@ -92,7 +92,7 @@ def test_symmetric_run_worker_node_behavior(monkeypatch, cleanup_ray):
 
                 # Test worker node behavior
                 args = ["--address", "192.168.1.100:6379", "--", "echo", "test"]
-                with patch("sys.argv", ["ray.scripts.symmetric_run", *args]):
+                with patch("sys.argv", ["/bin/ray", "symmetric-run", *args]):
                     with patch(
                         "ray.scripts.symmetric_run.check_head_node_ready"
                     ) as mock_check_head_node_ready:
@@ -134,7 +134,7 @@ def test_symmetric_run_arg_validation(monkeypatch, cleanup_ray):
             mock_run.return_value.returncode = 0
             args = ["--address", "127.0.0.1:6379", "--", "echo", "test"]
 
-            with patch("sys.argv", ["ray.scripts.symmetric_run", *args]):
+            with patch("sys.argv", ["/bin/ray", "symmetric-run", *args]):
                 # Test basic symmetric_run call using CliRunner
                 result = runner.invoke(symmetric_run, args)
                 assert result.exit_code == 0
@@ -144,7 +144,7 @@ def test_symmetric_run_arg_validation(monkeypatch, cleanup_ray):
             mock_run.return_value.returncode = 0
 
             args = ["--address", "127.0.0.1:6379", "echo", "test"]
-            with patch("sys.argv", ["ray.scripts.symmetric_run", *args]):
+            with patch("sys.argv", ["/bin/ray", "symmetric-run", *args]):
                 result = runner.invoke(symmetric_run, args)
                 assert result.exit_code == 1
                 assert "No separator" in result.output
@@ -154,7 +154,7 @@ def test_symmetric_run_arg_validation(monkeypatch, cleanup_ray):
             mock_run.return_value.returncode = 0
 
             args = ["--address", "127.0.0.1:6379", "--head", "--", "echo", "test"]
-            with patch("sys.argv", ["ray.scripts.symmetric_run", *args]):
+            with patch("sys.argv", ["/bin/ray", "symmetric-run", *args]):
                 result = runner.invoke(symmetric_run, args)
                 assert result.exit_code == 1
                 assert "Cannot use --head option in symmetric_run." in result.output
@@ -164,7 +164,7 @@ def test_symmetric_run_arg_validation(monkeypatch, cleanup_ray):
 
             # Test args with "=" are passed to ray start
             args = ["--address", "127.0.0.1:6379", "--num-cpus=4", "--", "echo", "test"]
-            with patch("sys.argv", ["ray.scripts.symmetric_run", *args]):
+            with patch("sys.argv", ["/bin/ray", "symmetric-run", *args]):
                 result = runner.invoke(symmetric_run, args)
                 assert result.exit_code == 0