Ensure skipped loop iteration register var is available (#83756)

Fixes #83619

changelogs/fragments/83619-loop-label-register.yml

@@ -0,0 +1,2 @@
+bugfixes:
+  - Fix an issue where registered variable was not available for templating in ``loop_control.label`` on skipped looped tasks (https://github.com/ansible/ansible/issues/83619)

lib/ansible/executor/task_executor.py

@@ -32,7 +32,7 @@ from ansible.utils.listify import listify_lookup_plugin_terms
 from ansible.utils.unsafe_proxy import to_unsafe_text, wrap_var
 from ansible.vars.clean import namespace_facts, clean_facts
 from ansible.utils.display import Display
-from ansible.utils.vars import combine_vars, isidentifier
+from ansible.utils.vars import combine_vars
 
 display = Display()
 
@@ -343,6 +343,13 @@ class TaskExecutor:
             (self._task, tmp_task) = (tmp_task, self._task)
             (self._play_context, tmp_play_context) = (tmp_play_context, self._play_context)
             res = self._execute(variables=task_vars)
+
+            if self._task.register:
+                # Ensure per loop iteration results are registered in case `_execute()`
+                # returns early (when conditional, failure, ...).
+                # This is needed in case the registered variable is used in the loop label template.
+                task_vars[self._task.register] = res
+
             task_fields = self._task.dump_attrs()
             (self._task, tmp_task) = (tmp_task, self._task)
             (self._play_context, tmp_play_context) = (tmp_play_context, self._play_context)
@@ -673,9 +680,6 @@ class TaskExecutor:
             # update the local copy of vars with the registered value, if specified,
             # or any facts which may have been generated by the module execution
             if self._task.register:
-                if not isidentifier(self._task.register):
-                    raise AnsibleError("Invalid variable name in 'register' specified: '%s'" % self._task.register)
-
                 vars_copy[self._task.register] = result
 
             if self._task.async_val > 0:

lib/ansible/playbook/task.py

@@ -37,6 +37,7 @@ from ansible.playbook.taggable import Taggable
 from ansible.utils.collection_loader import AnsibleCollectionConfig
 from ansible.utils.display import Display
 from ansible.utils.sentinel import Sentinel
+from ansible.utils.vars import isidentifier
 
 __all__ = ['Task']
 
@@ -274,6 +275,10 @@ class Task(Base, Conditional, Taggable, CollectionSearch, Notifiable, Delegatabl
         if not isinstance(value, list):
             setattr(self, name, [value])
 
+    def _validate_register(self, attr, name, value):
+        if value is not None and not isidentifier(value):
+            raise AnsibleParserError(f"Invalid variable name in 'register' specified: '{value}'")
+
     def post_validate(self, templar):
         '''
         Override of base class post_validate, to also do final validation on

lib/ansible/plugins/strategy/__init__.py

@@ -54,7 +54,7 @@ from ansible.utils.display import Display
 from ansible.utils.fqcn import add_internal_fqcns
 from ansible.utils.unsafe_proxy import wrap_var
 from ansible.utils.sentinel import Sentinel
-from ansible.utils.vars import combine_vars, isidentifier
+from ansible.utils.vars import combine_vars
 from ansible.vars.clean import strip_internal_keys, module_response_deepcopy
 
 display = Display()
@@ -766,10 +766,6 @@ class StrategyBase:
 
             # register final results
             if original_task.register:
-
-                if not isidentifier(original_task.register):
-                    raise AnsibleError("Invalid variable name in 'register' specified: '%s'" % original_task.register)
-
                 host_list = self.get_task_hosts(iterator, original_host, original_task)
 
                 clean_copy = strip_internal_keys(module_response_deepcopy(task_result._result))

test/integration/targets/loops/tasks/main.yml

@@ -405,3 +405,20 @@
 - assert:
     that:
       - foo[0] == 'foo1.0'
+
+# https://github.com/ansible/ansible/issues/83619
+- command: "echo {{ item }}"
+  register: r
+  loop:
+    - changed
+    - skipped
+  loop_control:
+    label: "{{ r.stdout }}"
+  when:
+    - item == "changed"
+  ignore_errors: true
+
+- name: test that the second iteration result was stored, since it was skipped no stdout should be present there
+  assert:
+    that:
+      - "r['results'][1]['msg'] is contains(\"no attribute 'stdout'\")"