diff --git a/lib/ansible/executor/task_executor.py b/lib/ansible/executor/task_executor.py
index 6fb45467e4d..28e67f4b6c5 100644
--- a/lib/ansible/executor/task_executor.py
+++ b/lib/ansible/executor/task_executor.py
@@ -279,6 +279,16 @@ class TaskExecutor:
             self._display.debug("when evaulation failed, skipping this task")
             return dict(changed=False, skipped=True, skip_reason='Conditional check failed', _ansible_no_log=self._play_context.no_log)
 
+        # if this task is a TaskInclude, we just return now with a success code so the
+        # main thread can expand the task list for the given host
+        if self._task.action == 'include':
+            include_file = self._task.args.pop('_raw_params', None)
+            if not include_file:
+                return dict(failed=True, msg="No include file was specified to the include")
+            else:
+                include_file = templar.template(include_file)
+            return dict(include=include_file, include_variables=self._task.args)
+
         # Now we do final validation on the task, which sets all fields to their final values.
         # In the case of debug tasks, we save any 'var' params and restore them after validating
         # so that variables are not replaced too early.
@@ -286,7 +296,6 @@ class TaskExecutor:
         if self._task.action == 'debug' and 'var' in self._task.args:
             prev_var = self._task.args.pop('var')
 
-        original_args = self._task.args.copy()
         self._task.post_validate(templar=templar)
         if '_variable_params' in self._task.args:
             variable_params = self._task.args.pop('_variable_params')
@@ -298,14 +307,6 @@ class TaskExecutor:
         if prev_var is not None:
             self._task.args['var'] = prev_var
 
-        # if this task is a TaskInclude, we just return now with a success code so the
-        # main thread can expand the task list for the given host
-        if self._task.action == 'include':
-            include_variables = original_args
-            include_file = include_variables.get('_raw_params')
-            del include_variables['_raw_params']
-            return dict(include=include_file, include_variables=include_variables)
-
         # get the connection and the handler for this execution
         self._connection = self._get_connection(variables=variables, templar=templar)
         self._connection.set_host_overrides(host=self._host)
diff --git a/lib/ansible/plugins/strategy/__init__.py b/lib/ansible/plugins/strategy/__init__.py
index 4aa42d304fb..ca5d7b96720 100644
--- a/lib/ansible/plugins/strategy/__init__.py
+++ b/lib/ansible/plugins/strategy/__init__.py
@@ -401,6 +401,7 @@ class StrategyBase:
 
         try:
             data = self._loader.load_from_file(included_file._filename)
+            self._tqm.send_callback('v2_playbook_on_include', included_file)
             if data is None:
                 return []
         except AnsibleError as e: