diff --git a/changelogs/fragments/delegate_has_hostvars.yml b/changelogs/fragments/delegate_has_hostvars.yml
new file mode 100644
index 00000000000..9e3dd93ae8f
--- /dev/null
+++ b/changelogs/fragments/delegate_has_hostvars.yml
@@ -0,0 +1,4 @@
+bugfixes:
+  - ensure delegated vars can resolve hostvars object and access vars from hostvars[inventory_hostname].
+  - fix issue with inventory_hostname and delegated host vars mixing on connection settings.
+  - add magic/connection vars updates from delegated host info.
diff --git a/lib/ansible/executor/task_executor.py b/lib/ansible/executor/task_executor.py
index ba15257ca9d..1c87a2148cf 100644
--- a/lib/ansible/executor/task_executor.py
+++ b/lib/ansible/executor/task_executor.py
@@ -38,6 +38,8 @@ from ansible.utils.vars import combine_vars, isidentifier
 display = Display()
 
 
+RETURN_VARS = [x for x in C.MAGIC_VARIABLE_MAPPING.items() if 'become' not in x and '_pass' not in x]
+
 __all__ = ['TaskExecutor']
 
 
@@ -510,6 +512,10 @@ class TaskExecutor:
 
         context_validation_error = None
         try:
+            # TODO: remove play_context as this does not take delegation into account, task itself should hold values
+            #  for connection/shell/become/terminal plugin options to finalize.
+            #  Kept for now for backwards compatiblity and a few functions that are still exclusive to it.
+
             # apply the given task's information to the connection info,
             # which may override some fields already set by the play or
             # the options specified on the command line
@@ -528,7 +534,6 @@ class TaskExecutor:
             # a certain subset of variables exist.
             self._play_context.update_vars(variables)
 
-            # FIXME: update connection/shell plugin options
         except AnsibleError as e:
             # save the error, which we'll raise later if we don't end up
             # skipping this task during the conditional evaluation step
@@ -591,26 +596,28 @@ class TaskExecutor:
                 variable_params.update(self._task.args)
                 self._task.args = variable_params
 
+        if self._task.delegate_to:
+            # use vars from delegated host (which already include task vars) instead of original host
+            cvars = variables.get('ansible_delegated_vars', {}).get(self._task.delegate_to, {})
+            orig_vars = templar.available_variables
+        else:
+            # just use normal host vars
+            cvars = orig_vars = variables
+
+        templar.available_variables = cvars
+
         # get the connection and the handler for this execution
         if (not self._connection or
                 not getattr(self._connection, 'connected', False) or
                 self._play_context.remote_addr != self._connection._play_context.remote_addr):
-            self._connection = self._get_connection(variables=variables, templar=templar)
+            self._connection = self._get_connection(cvars, templar)
         else:
             # if connection is reused, its _play_context is no longer valid and needs
             # to be replaced with the one templated above, in case other data changed
             self._connection._play_context = self._play_context
 
-        if self._task.delegate_to:
-            # use vars from delegated host (which already include task vars) instead of original host
-            delegated_vars = variables.get('ansible_delegated_vars', {}).get(self._task.delegate_to, {})
-            orig_vars = templar.available_variables
-            templar.available_variables = delegated_vars
-            plugin_vars = self._set_connection_options(delegated_vars, templar)
-            templar.available_variables = orig_vars
-        else:
-            # just use normal host vars
-            plugin_vars = self._set_connection_options(variables, templar)
+        plugin_vars = self._set_connection_options(cvars, templar)
+        templar.available_variables = orig_vars
 
         # get handler
         self._handler = self._get_action_handler(connection=self._connection, templar=templar)
@@ -788,10 +795,17 @@ class TaskExecutor:
 
         # add the delegated vars to the result, so we can reference them
         # on the results side without having to do any further templating
+        # also now add conneciton vars results when delegating
         if self._task.delegate_to:
             result["_ansible_delegated_vars"] = {'ansible_delegated_host': self._task.delegate_to}
-            for k in plugin_vars:
-                result["_ansible_delegated_vars"][k] = delegated_vars.get(k)
+            for k in plugin_vars + RETURN_VARS:
+                if k in cvars and cvars[k] is not None:
+                    result["_ansible_delegated_vars"][k] = cvars[k]
+        else:
+            for k in plugin_vars + RETURN_VARS:
+                if k in cvars and cvars[k] is not None:
+                    result[k] = cvars[k]
+
         # and return
         display.debug("attempt loop complete, returning result")
         return result
@@ -877,17 +891,12 @@ class TaskExecutor:
                                "Use `ansible-doc -t become -l` to list available plugins." % name)
         return become
 
-    def _get_connection(self, variables, templar):
+    def _get_connection(self, cvars, templar):
         '''
         Reads the connection property for the host, and returns the
         correct connection object from the list of connection plugins
         '''
 
-        if self._task.delegate_to is not None:
-            cvars = variables.get('ansible_delegated_vars', {}).get(self._task.delegate_to, {})
-        else:
-            cvars = variables
-
         # use magic var if it exists, if not, let task inheritance do it's thing.
         if cvars.get('ansible_connection') is not None:
             self._play_context.connection = templar.template(cvars['ansible_connection'])
@@ -949,15 +958,14 @@ class TaskExecutor:
             display.vvvv('attempting to start connection', host=self._play_context.remote_addr)
             display.vvvv('using connection plugin %s' % connection.transport, host=self._play_context.remote_addr)
 
-            options = self._get_persistent_connection_options(connection, variables, templar)
+            options = self._get_persistent_connection_options(connection, cvars, templar)
             socket_path = start_connection(self._play_context, options, self._task._uuid)
             display.vvvv('local domain socket path is %s' % socket_path, host=self._play_context.remote_addr)
             setattr(connection, '_socket_path', socket_path)
 
         return connection
 
-    def _get_persistent_connection_options(self, connection, variables, templar):
-        final_vars = combine_vars(variables, variables.get('ansible_delegated_vars', dict()).get(self._task.delegate_to, dict()))
+    def _get_persistent_connection_options(self, connection, final_vars, templar):
 
         option_vars = C.config.get_plugin_vars('connection', connection._load_name)
         plugin = connection._sub_plugin
diff --git a/lib/ansible/vars/manager.py b/lib/ansible/vars/manager.py
index be4954c5b16..a4942bac5d3 100644
--- a/lib/ansible/vars/manager.py
+++ b/lib/ansible/vars/manager.py
@@ -602,8 +602,9 @@ class VariableManager:
                 host=delegated_host,
                 task=task,
                 include_delegate_to=False,
-                include_hostvars=False,
+                include_hostvars=True,
             )
+            delegated_host_vars[delegated_host_name]['inventory_hostname'] = vars_copy.get('inventory_hostname')
 
         _ansible_loop_cache = None
         if has_loop and cache_items:
diff --git a/test/integration/targets/delegate_to/connection_plugins/fakelocal.py b/test/integration/targets/delegate_to/connection_plugins/fakelocal.py
new file mode 100644
index 00000000000..59ddcf05eee
--- /dev/null
+++ b/test/integration/targets/delegate_to/connection_plugins/fakelocal.py
@@ -0,0 +1,76 @@
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+DOCUMENTATION = '''
+    connection: fakelocal
+    short_description: dont execute anything
+    description:
+        - This connection plugin just verifies parameters passed in
+    author: ansible (@core)
+    version_added: histerical
+    options:
+      password:
+          description: Authentication password for the C(remote_user). Can be supplied as CLI option.
+          vars:
+              - name: ansible_password
+      remote_user:
+          description:
+              - User name with which to login to the remote server, normally set by the remote_user keyword.
+          ini:
+            - section: defaults
+              key: remote_user
+          vars:
+              - name: ansible_user
+'''
+
+from ansible.errors import AnsibleConnectionFailure
+from ansible.plugins.connection import ConnectionBase
+from ansible.utils.display import Display
+
+display = Display()
+
+
+class Connection(ConnectionBase):
+    ''' Local based connections '''
+
+    transport = 'fakelocal'
+    has_pipelining = True
+
+    def __init__(self, *args, **kwargs):
+
+        super(Connection, self).__init__(*args, **kwargs)
+        self.cwd = None
+
+    def _connect(self):
+        ''' verify '''
+
+        if self.get_option('remote_user') == 'invaliduser' and self.get_option('password') == 'badpassword':
+            raise AnsibleConnectionFailure('Got invaliduser and badpassword')
+
+        if not self._connected:
+            display.vvv(u"ESTABLISH FAKELOCAL CONNECTION FOR USER: {0}".format(self._play_context.remote_user), host=self._play_context.remote_addr)
+            self._connected = True
+        return self
+
+    def exec_command(self, cmd, in_data=None, sudoable=True):
+        ''' run a command on the local host '''
+
+        super(Connection, self).exec_command(cmd, in_data=in_data, sudoable=sudoable)
+
+        return 0, '{"msg": "ALL IS GOOD"}', ''
+
+    def put_file(self, in_path, out_path):
+        ''' transfer a file from local to local '''
+
+        super(Connection, self).put_file(in_path, out_path)
+
+    def fetch_file(self, in_path, out_path):
+        ''' fetch a file from local to local -- for compatibility '''
+
+        super(Connection, self).fetch_file(in_path, out_path)
+
+    def close(self):
+        ''' terminate the connection; nothing to do here '''
+        self._connected = False
diff --git a/test/integration/targets/delegate_to/has_hostvars.yml b/test/integration/targets/delegate_to/has_hostvars.yml
new file mode 100644
index 00000000000..9e8926bde8a
--- /dev/null
+++ b/test/integration/targets/delegate_to/has_hostvars.yml
@@ -0,0 +1,64 @@
+- name: ensure delegated host has hostvars available for resolving connection
+  hosts: testhost
+  gather_facts: false
+  tasks:
+
+    - name: ensure delegated host uses current host as inventory_hostname
+      assert:
+        that:
+          - inventory_hostname == ansible_delegated_vars['testhost5']['inventory_hostname']
+      delegate_to: testhost5
+
+    - name: Set info on inventory_hostname
+      set_fact:
+        login: invaliduser
+        mypass: badpassword
+
+    - name: test fakelocal
+      command: ls
+      ignore_unreachable: True
+      ignore_errors: True
+      remote_user: "{{ login }}"
+      vars:
+        ansible_password: "{{ mypass }}"
+        ansible_connection: fakelocal
+      register: badlogin
+
+    - name: ensure we skipped do to unreachable and not templating error
+      assert:
+        that:
+          - badlogin is unreachable
+
+    - name: delegate but try to use inventory_hostname data directly
+      command: ls
+      delegate_to: testhost5
+      ignore_unreachable: True
+      ignore_errors: True
+      remote_user: "{{ login }}"
+      vars:
+        ansible_password: "{{ mypass }}"
+      register: badlogin
+
+    - name: ensure we skipped do to unreachable and not templating error
+      assert:
+        that:
+          - badlogin is not unreachable
+          - badlogin is failed
+          - "'undefined' in badlogin['msg']"
+
+    - name: delegate ls to testhost5 as it uses ssh while testhost is local, but use vars from testhost
+      command: ls
+      remote_user: "{{ hostvars[inventory_hostname]['login'] }}"
+      delegate_to: testhost5
+      ignore_unreachable: True
+      ignore_errors: True
+      vars:
+        ansible_password: "{{ hostvars[inventory_hostname]['mypass'] }}"
+      register: badlogin
+
+    - name: ensure we skipped do to unreachable and not templating error
+      assert:
+        that:
+          - badlogin is unreachable
+          - badlogin is not failed
+          - "'undefined' not in badlogin['msg']"
diff --git a/test/integration/targets/delegate_to/inventory b/test/integration/targets/delegate_to/inventory
index 8a2157e2bb0..f7ad0a33dad 100644
--- a/test/integration/targets/delegate_to/inventory
+++ b/test/integration/targets/delegate_to/inventory
@@ -3,6 +3,7 @@ testhost ansible_connection=local
 testhost2 ansible_connection=local
 testhost3 ansible_ssh_host=127.0.0.3
 testhost4 ansible_ssh_host=127.0.0.4
+testhost5 ansible_connection=fakelocal
 
 [all:vars]
 ansible_python_interpreter="{{ ansible_playbook_python }}"
diff --git a/test/integration/targets/delegate_to/runme.sh b/test/integration/targets/delegate_to/runme.sh
index 8095c921d23..5e08f5c36a7 100755
--- a/test/integration/targets/delegate_to/runme.sh
+++ b/test/integration/targets/delegate_to/runme.sh
@@ -59,6 +59,7 @@ ansible-playbook test_delegate_to_loop_caching.yml -i inventory -v "$@"
 # ensure we are using correct settings when delegating
 ANSIBLE_TIMEOUT=3 ansible-playbook delegate_vars_hanldling.yml -i inventory -v "$@"
 
+ansible-playbook has_hostvars.yml -i inventory -v "$@"
 
 # test ansible_x_interpreter
 # python
diff --git a/test/integration/targets/delegate_to/test_delegate_to.yml b/test/integration/targets/delegate_to/test_delegate_to.yml
index d75857cc6cf..05b0536e680 100644
--- a/test/integration/targets/delegate_to/test_delegate_to.yml
+++ b/test/integration/targets/delegate_to/test_delegate_to.yml
@@ -18,6 +18,8 @@
       register: setup_results
       delegate_to: testhost4
 
+    - debug: var=setup_results
+
     - assert:
         that:
           - '"127.0.0.4" in setup_results.ansible_facts.ansible_env["SSH_CONNECTION"]'