fix(cli): only treat delimited run/executor/plugins prefixes as overwrites

nemo_run/cli/api.py

@@ -1659,19 +1659,22 @@ def _parse_prefixed_args(
     """
     prefixed_arg_value, prefixed_args, other_args = None, [], []
     for arg in args:
-        if arg.startswith(prefix):
-            if arg.startswith(f"{prefix}="):
-                prefixed_arg_value = arg.split("=")[1]
-            else:
-                if not arg.startswith(f"{prefix}.") and not arg.startswith(f"{prefix}["):
-                    raise ValueError(
-                        f"{prefix.capitalize()} overwrites must start with '{prefix}.'. Got {arg}"
-                    )
-                if arg.startswith(f"{prefix}."):
-                    prefixed_args.append(arg.replace(f"{prefix}.", ""))
-                elif arg.startswith(f"{prefix}["):
-                    prefixed_args.append(arg.replace(prefix, ""))
+        if arg.startswith(f"{prefix}="):
+            prefixed_arg_value = arg.split("=", 1)[1]
+        elif arg.startswith(f"{prefix}."):
+            prefixed_args.append(arg.replace(f"{prefix}.", "", 1))
+        elif arg.startswith(f"{prefix}["):
+            prefixed_args.append(arg.replace(prefix, "", 1))
+        elif arg.startswith(prefix) and "=" not in arg:
+            # A bare token starting with the prefix (e.g. a positional value)
+            # cannot address a task parameter, so treat it as a malformed overwrite.
+            raise ValueError(
+                f"{prefix.capitalize()} overwrites must start with '{prefix}.'. Got {arg}"
+            )
         else:
+            # Keyword arguments for parameters whose names merely start with the
+            # prefix (e.g. runtime=3600 for prefix "run") belong to the task,
+            # not to the prefixed namespace.
             other_args.append(arg)
     return prefixed_arg_value, prefixed_args, other_args
 

test/cli/test_api.py

@@ -167,6 +167,22 @@ def test_run_context_execute_task(self, mock_run, mock_dryrun_fn, sample_functio
         mock_dryrun_fn.assert_called_once()
         mock_run.assert_called_once()
 
+    @patch("nemo_run.dryrun_fn")
+    @patch("nemo_run.run")
+    def test_run_context_execute_task_with_run_prefixed_parameter(
+        self, mock_run, mock_dryrun_fn
+    ):
+        """Task parameters whose names start with a reserved prefix (run/executor/plugins)
+        must be treated as task args, not as malformed prefixed overwrites."""
+        ctx = RunContext(name="test_run", skip_confirmation=True)
+
+        def sample_function(runtime: int = 60, lr: float = 0.1):
+            return None
+
+        ctx.cli_execute(sample_function, ["runtime=120", "lr=0.2"])
+        mock_dryrun_fn.assert_called_once()
+        mock_run.assert_called_once()
+
     def test_run_context_to_config(self):
         ctx = RunContext(name="test_run")
         config = ctx.to_config()
@@ -1163,6 +1179,51 @@ def test_parse_prefixed_args_no_prefix(self):
         assert prefix_args == []
         assert other_args == ["arg1=value1", "arg2=value2"]
 
+    def test_parse_prefixed_args_word_boundary(self):
+        """Keyword args whose parameter names merely start with the prefix are task args."""
+        from nemo_run.cli.api import _parse_prefixed_args
+
+        args = ["runtime=3600", "run_id=42", "lr=0.1"]
+        prefix_value, prefix_args, other_args = _parse_prefixed_args(args, "run")
+
+        assert prefix_value is None
+        assert prefix_args == []
+        assert other_args == args
+
+        prefix_value, prefix_args, other_args = _parse_prefixed_args(["executors=2"], "executor")
+        assert prefix_value is None
+        assert prefix_args == []
+        assert other_args == ["executors=2"]
+
+        prefix_value, prefix_args, other_args = _parse_prefixed_args(
+            ["plugins_dir=/tmp"], "plugins"
+        )
+        assert prefix_value is None
+        assert prefix_args == []
+        assert other_args == ["plugins_dir=/tmp"]
+
+        # Nested keys starting with the prefix stay task args as well.
+        _, prefix_args, other_args = _parse_prefixed_args(["runtime.limit=60"], "run")
+        assert prefix_args == []
+        assert other_args == ["runtime.limit=60"]
+
+    def test_parse_prefixed_args_nested_key_not_corrupted(self):
+        """Only the leading prefix is stripped from prefixed args."""
+        from nemo_run.cli.api import _parse_prefixed_args
+
+        _, prefix_args, _ = _parse_prefixed_args(["run.a.run.b=1"], "run")
+        assert prefix_args == ["a.run.b=1"]
+
+        _, prefix_args, _ = _parse_prefixed_args(["plugins[0].plugins_dir=x"], "plugins")
+        assert prefix_args == ["[0].plugins_dir=x"]
+
+    def test_parse_prefixed_args_value_with_equals(self):
+        """Values containing '=' are not truncated at the first '='."""
+        from nemo_run.cli.api import _parse_prefixed_args
+
+        prefix_value, _, _ = _parse_prefixed_args(["executor=k=v"], "executor")
+        assert prefix_value == "k=v"
+
 
 class TestConfigExport:
     @pytest.fixture