Validate input args in luigi.run()

luigi/interface.py

@@ -159,6 +159,13 @@ def run(*args, **kwargs):
 
     :param use_dynamic_argparse: Deprecated and ignored
     """
+    cmdline_args = args[0] if args else kwargs.get("cmdline_args")
+    if cmdline_args is not None and not isinstance(cmdline_args, (list, tuple)):
+        raise TypeError(
+            f"luigi.run() expects a list or tuple of command-line arguments, "
+            f"got {type(cmdline_args).__name__!r}. "
+            "Example: luigi.run(['MyTask', '--param', 'value'])"
+        )
     luigi_run_result = _run(*args, **kwargs)
     return luigi_run_result if kwargs.get("detailed_summary") else luigi_run_result.scheduling_succeeded
 

test/interface_test.py

@@ -178,6 +178,34 @@ class MyOtherTestTask(luigi.Task):
         with patch.object(sys, "argv", ["my_module.py", "--no-lock", "--my-param", "my_value", "--local-scheduler"]):
             luigi.run(main_task_cls=MyOtherTestTask)
 
+    def test_run_with_int_raises_type_error(self):
+        with self.assertRaises(TypeError) as ctx:
+            luigi.run(123)
+        self.assertIn("int", str(ctx.exception))
+        self.assertIn("list or tuple", str(ctx.exception))
+
+    def test_run_with_dict_raises_type_error(self):
+        with self.assertRaises(TypeError) as ctx:
+            luigi.run({"task": "MyTask"})
+        self.assertIn("dict", str(ctx.exception))
+        self.assertIn("list or tuple", str(ctx.exception))
+
+    def test_run_with_none_does_not_raise_type_error(self):
+        try:
+            luigi.run(None)
+        except TypeError:
+            self.fail("luigi.run() raised TypeError for None argument")
+        except (Exception, SystemExit):
+            pass  # scheduler/task errors and SystemExit are fine
+
+    def test_run_with_valid_list_does_not_raise_type_error(self):
+        try:
+            luigi.run(["MyTask"])
+        except TypeError:
+            self.fail("luigi.run() raised TypeError for a valid list argument")
+        except Exception:
+            pass  # scheduler/task errors are fine
+
     def _run_interface(self, **env_params):
         return luigi.interface.build([self.task_a, self.task_b], worker_scheduler_factory=self.worker_scheduler_factory, **env_params)