diff --git a/airflow/executors/base_executor.py b/airflow/executors/base_executor.py
index 8361ba4bef27a..33a666b550580 100644
--- a/airflow/executors/base_executor.py
+++ b/airflow/executors/base_executor.py
@@ -363,7 +363,7 @@ def validate_airflow_tasks_run_command(command: list[str]) -> tuple[str | None,
         if len(command) > 3 and "--help" not in command:
             dag_id: str | None = None
             task_id: str | None = None
-            for arg in command[4:]:
+            for arg in command[3:]:
                 if not arg.startswith("--"):
                     if dag_id is None:
                         dag_id = arg
diff --git a/tests/executors/test_base_executor.py b/tests/executors/test_base_executor.py
index e4e39e45a0fdc..25f7235479941 100644
--- a/tests/executors/test_base_executor.py
+++ b/tests/executors/test_base_executor.py
@@ -125,3 +125,10 @@ def test_trigger_running_tasks(dag_maker, change_state_attempt):
         assert len(executor.execute_async.mock_calls) == len(dagrun.task_instances) + 1
     else:
         assert len(executor.execute_async.mock_calls) == len(dagrun.task_instances)
+
+
+def test_validate_airflow_tasks_run_command(dag_maker):
+    dagrun = setup_dagrun(dag_maker)
+    tis = dagrun.task_instances
+    dag_id, task_id = BaseExecutor.validate_airflow_tasks_run_command(tis[0].command_as_list())
+    assert dag_id == dagrun.dag_id and task_id == tis[0].task_id