diff --git a/lib/ansible/playbook/__init__.py b/lib/ansible/playbook/__init__.py
index 58e2bafe18e..28e1d923eb8 100644
--- a/lib/ansible/playbook/__init__.py
+++ b/lib/ansible/playbook/__init__.py
@@ -399,6 +399,9 @@ class PlayBook(object):
             remote_user=task.remote_user,
             remote_port=task.play.remote_port,
             module_vars=task.module_vars,
+            play_vars=task.play_vars,
+            play_file_vars=task.play_file_vars,
+            role_vars=task.role_vars,
             default_vars=task.default_vars,
             extra_vars=self.extra_vars,
             private_key_file=self.private_key_file,
@@ -500,7 +503,7 @@ class PlayBook(object):
         def _save_play_facts(host, facts):
             # saves play facts in SETUP_CACHE, unless the module executed was
             # set_fact, in which case we add them to the VARS_CACHE
-            if task.module_name == 'set_fact':
+            if task.module_name in ('set_fact', 'include_vars'):
                 utils.update_hash(self.VARS_CACHE, host, facts)
             else:
                 utils.update_hash(self.SETUP_CACHE, host, facts)
@@ -605,6 +608,9 @@ class PlayBook(object):
             transport=play.transport,
             is_playbook=True,
             module_vars=play.vars,
+            play_vars=play.vars,
+            play_file_vars=play.vars_file_vars,
+            role_vars=play.role_vars,
             default_vars=play.default_vars,
             check=self.check,
             diff=self.diff,
diff --git a/lib/ansible/playbook/play.py b/lib/ansible/playbook/play.py
index 742c12b382d..b793247826a 100644
--- a/lib/ansible/playbook/play.py
+++ b/lib/ansible/playbook/play.py
@@ -33,7 +33,7 @@ import uuid
 class Play(object):
 
     __slots__ = [
-       'hosts', 'name', 'vars', 'default_vars', 'vars_prompt', 'vars_files',
+       'hosts', 'name', 'vars', 'vars_file_vars', 'role_vars', 'default_vars', 'vars_prompt', 'vars_files',
        'handlers', 'remote_user', 'remote_port', 'included_roles', 'accelerate',
        'accelerate_port', 'accelerate_ipv6', 'sudo', 'sudo_user', 'transport', 'playbook',
        'tags', 'gather_facts', 'serial', '_ds', '_handlers', '_tasks',
@@ -65,6 +65,8 @@ class Play(object):
         self.vars_prompt      = ds.get('vars_prompt', {})
         self.playbook         = playbook
         self.vars             = self._get_vars()
+        self.vars_file_vars   = dict() # these are vars read in from vars_files:
+        self.role_vars        = dict() # these are vars read in from vars/main.yml files in roles
         self.basedir          = basedir
         self.roles            = ds.get('roles', None)
         self.tags             = ds.get('tags', None)
@@ -108,10 +110,6 @@ class Play(object):
 
         self._update_vars_files_for_host(None)
 
-        # apply any extra_vars specified on the command line now
-        if type(self.playbook.extra_vars) == dict:
-            self.vars = utils.combine_vars(self.vars, self.playbook.extra_vars)
-
         # template everything to be efficient, but do not pre-mature template
         # tasks/handlers as they may have inventory scope overrides
         _tasks    = ds.pop('tasks', [])
@@ -224,6 +222,7 @@ class Play(object):
         for role in roles:
             role_path,role_vars = self._get_role_path(role)
             role_vars = utils.combine_vars(passed_vars, role_vars)
+
             vars = self._resolve_main(utils.path_dwim(self.basedir, os.path.join(role_path, 'vars')))
             vars_data = {}
             if os.path.isfile(vars):
@@ -232,10 +231,12 @@ class Play(object):
                     if not isinstance(vars_data, dict):
                         raise errors.AnsibleError("vars from '%s' are not a dict" % vars)
                     role_vars = utils.combine_vars(vars_data, role_vars)
+
             defaults = self._resolve_main(utils.path_dwim(self.basedir, os.path.join(role_path, 'defaults')))
             defaults_data = {}
             if os.path.isfile(defaults):
                 defaults_data = utils.parse_yaml_from_file(defaults, vault_password=self.vault_password)
+
             # the meta directory contains the yaml that should
             # hold the list of dependencies (if any)
             meta = self._resolve_main(utils.path_dwim(self.basedir, os.path.join(role_path, 'meta')))
@@ -287,13 +288,15 @@ class Play(object):
 
                         dep_vars = utils.combine_vars(passed_vars, dep_vars)
                         dep_vars = utils.combine_vars(role_vars, dep_vars)
+
                         vars = self._resolve_main(utils.path_dwim(self.basedir, os.path.join(dep_path, 'vars')))
                         vars_data = {}
                         if os.path.isfile(vars):
                             vars_data = utils.parse_yaml_from_file(vars, vault_password=self.vault_password)
                             if vars_data:
-                                #dep_vars = utils.combine_vars(vars_data, dep_vars)
                                 dep_vars = utils.combine_vars(dep_vars, vars_data)
+                                pass
+
                         defaults = self._resolve_main(utils.path_dwim(self.basedir, os.path.join(dep_path, 'defaults')))
                         dep_defaults_data = {}
                         if os.path.isfile(defaults):
@@ -338,6 +341,19 @@ class Play(object):
                 dep_stack.append([role,role_path,role_vars,defaults_data])
         return dep_stack
 
+    def _load_role_vars_files(self, vars_files):
+        # process variables stored in vars/main.yml files
+        role_vars = {}
+        for filename in vars_files:
+            if os.path.exists(filename):
+                new_vars = utils.parse_yaml_from_file(filename, vault_password=self.vault_password)
+                if new_vars:
+                    if type(new_vars) != dict:
+                        raise errors.AnsibleError("%s must be stored as dictionary/hash: %s" % (filename, type(new_vars)))
+                    role_vars = utils.combine_vars(role_vars, new_vars)
+
+        return role_vars
+
     def _load_role_defaults(self, defaults_files):
         # process default variables
         default_vars = {}
@@ -364,10 +380,10 @@ class Play(object):
         if type(roles) != list:
             raise errors.AnsibleError("value of 'roles:' must be a list")
 
-        new_tasks = []
-        new_handlers = []
-        new_vars_files = []
-        defaults_files = []
+        new_tasks       = []
+        new_handlers    = []
+        role_vars_files = []
+        defaults_files  = []
 
         pre_tasks = ds.get('pre_tasks', None)
         if type(pre_tasks) != list:
@@ -434,7 +450,7 @@ class Play(object):
                         nt[k] = special_vars[k]
                 new_handlers.append(nt)
             if os.path.isfile(vars_file):
-                new_vars_files.append(vars_file)
+                role_vars_files.append(vars_file)
             if os.path.isfile(defaults_file):
                 defaults_files.append(defaults_file)
             if os.path.isdir(library):
@@ -462,13 +478,12 @@ class Play(object):
         new_tasks.append(dict(meta='flush_handlers'))
 
         new_handlers.extend(handlers)
-        new_vars_files.extend(vars_files)
 
         ds['tasks'] = new_tasks
         ds['handlers'] = new_handlers
-        ds['vars_files'] = new_vars_files
         ds['role_names'] = role_names
 
+        self.role_vars = self._load_role_vars_files(role_vars_files)
         self.default_vars = self._load_role_defaults(defaults_files)
 
         return ds
@@ -535,8 +550,7 @@ class Play(object):
                     results.append(Task(self, x))
                     continue
 
-            task_vars = self.vars.copy()
-            task_vars.update(vars)
+            task_vars = vars.copy()
             if original_file:
                 task_vars['_original_file'] = original_file
 
@@ -601,6 +615,9 @@ class Play(object):
                 task = Task(
                     self, x,
                     module_vars=task_vars,
+                    play_vars=self.vars,
+                    play_file_vars=self.vars_file_vars,
+                    role_vars=self.role_vars,
                     default_vars=default_vars,
                     additional_conditions=list(additional_conditions),
                     role_name=role_name
@@ -818,7 +835,7 @@ class Play(object):
                             target_filename = filename4
                     update_vars_cache(host, data, target_filename=target_filename)
                 else:
-                    self.vars = utils.combine_vars(self.vars, data)
+                    self.vars_file_vars = utils.combine_vars(self.vars_file_vars, data)
                 # we did process this file
                 return True
             # we did not process this file
diff --git a/lib/ansible/playbook/task.py b/lib/ansible/playbook/task.py
index db10f7c494d..ebe43f63c1c 100644
--- a/lib/ansible/playbook/task.py
+++ b/lib/ansible/playbook/task.py
@@ -26,7 +26,7 @@ class Task(object):
 
     __slots__ = [
         'name', 'meta', 'action', 'when', 'async_seconds', 'async_poll_interval',
-        'notify', 'module_name', 'module_args', 'module_vars', 'default_vars',
+        'notify', 'module_name', 'module_args', 'module_vars', 'play_vars', 'play_file_vars', 'role_vars', 'default_vars',
         'play', 'notified_by', 'tags', 'register', 'role_name',
         'delegate_to', 'first_available_file', 'ignore_errors',
         'local_action', 'transport', 'sudo', 'remote_user', 'sudo_user', 'sudo_pass',
@@ -45,7 +45,7 @@ class Task(object):
          'su', 'su_user', 'su_pass', 'no_log', 'run_once',
     ]
 
-    def __init__(self, play, ds, module_vars=None, default_vars=None, additional_conditions=None, role_name=None):
+    def __init__(self, play, ds, module_vars=None, play_vars=None, play_file_vars=None, role_vars=None, default_vars=None, additional_conditions=None, role_name=None):
         ''' constructor loads from a task or handler datastructure '''
 
         # meta directives are used to tell things like ansible/playbook to run
@@ -119,9 +119,12 @@ class Task(object):
             elif not x in Task.VALID_KEYS:
                 raise errors.AnsibleError("%s is not a legal parameter in an Ansible task or handler" % x)
 
-        self.module_vars  = module_vars
-        self.default_vars = default_vars
-        self.play         = play
+        self.module_vars    = module_vars
+        self.play_vars      = play_vars
+        self.play_file_vars = play_file_vars
+        self.role_vars      = role_vars
+        self.default_vars   = default_vars
+        self.play           = play
 
         # load various attributes
         self.name         = ds.get('name', None)
@@ -219,6 +222,9 @@ class Task(object):
 
         # combine the default and module vars here for use in templating
         all_vars = self.default_vars.copy()
+        all_vars = utils.combine_vars(all_vars, self.play_vars)
+        all_vars = utils.combine_vars(all_vars, self.play_file_vars)
+        all_vars = utils.combine_vars(all_vars, self.role_vars)
         all_vars = utils.combine_vars(all_vars, self.module_vars)
 
         self.async_seconds = ds.get('async', 0)  # not async by default
diff --git a/lib/ansible/runner/__init__.py b/lib/ansible/runner/__init__.py
index 8f271f0500f..1d236f5f11e 100644
--- a/lib/ansible/runner/__init__.py
+++ b/lib/ansible/runner/__init__.py
@@ -134,7 +134,10 @@ class Runner(object):
         sudo=False,                         # whether to run sudo or not
         sudo_user=C.DEFAULT_SUDO_USER,      # ex: 'root'
         module_vars=None,                   # a playbooks internals thing
-        default_vars=None,                  # ditto
+        play_vars=None,                     #
+        play_file_vars=None,                #
+        role_vars=None,                     #
+        default_vars=None,                  #
         extra_vars=None,                    # extra vars specified with he playbook(s)
         is_playbook=False,                  # running from playbook or not?
         inventory=None,                     # reference to Inventory object
@@ -176,6 +179,9 @@ class Runner(object):
         self.inventory        = utils.default(inventory, lambda: ansible.inventory.Inventory(host_list))
 
         self.module_vars      = utils.default(module_vars, lambda: {})
+        self.play_vars        = utils.default(play_vars, lambda: {})
+        self.play_file_vars   = utils.default(play_file_vars, lambda: {})
+        self.role_vars        = utils.default(role_vars, lambda: {})
         self.default_vars     = utils.default(default_vars, lambda: {})
         self.extra_vars       = utils.default(extra_vars, lambda: {})
 
@@ -629,10 +635,16 @@ class Runner(object):
         inject = utils.combine_vars(inject, host_variables)
         # then the setup_cache which contains facts gathered
         inject = utils.combine_vars(inject, self.setup_cache.get(host, {}))
-        # followed by vars (vars, vars_files, vars/main.yml)
-        inject = utils.combine_vars(inject, self.vars_cache.get(host, {}))
+        # next come variables from vars and vars files
+        inject = utils.combine_vars(inject, self.play_vars)
+        inject = utils.combine_vars(inject, self.play_file_vars)
+        # next come variables from role vars/main.yml files
+        inject = utils.combine_vars(inject, self.role_vars)
         # then come the module variables
         inject = utils.combine_vars(inject, module_vars)
+        # followed by vars_cache things (set_fact, include_vars, and
+        # vars_files which had host-specific templating done)
+        inject = utils.combine_vars(inject, self.vars_cache.get(host, {}))
         # and finally -e vars are the highest priority
         inject = utils.combine_vars(inject, self.extra_vars)
         # and then special vars
diff --git a/test/integration/Makefile b/test/integration/Makefile
index 6568c530171..b03c3eff786 100644
--- a/test/integration/Makefile
+++ b/test/integration/Makefile
@@ -19,7 +19,7 @@ TMPDIR = $(shell mktemp -d 2>/dev/null || mktemp -d -t 'mytmpdir')
 
 VAULT_PASSWORD_FILE = vault-password
 
-all: non_destructive destructive includes unicode test_var_precedence check_mode test_hash test_handlers test_group_by test_vault parsing
+all: parsing test_var_precedence unicode non_destructive destructive includes check_mode test_hash test_handlers test_group_by test_vault
 
 parsing:
 	ansible-playbook bad_parsing.yml -i $(INVENTORY) -e @$(VARS_FILE) $(CREDENTIALS_ARG) -vvv $(TEST_FLAGS) --tags common,scenario1; [ $$? -eq 3 ]
diff --git a/test/integration/roles/test_var_precedence/tasks/main.yml b/test/integration/roles/test_var_precedence/tasks/main.yml
index 1915ebdb916..7850e6b6463 100644
--- a/test/integration/roles/test_var_precedence/tasks/main.yml
+++ b/test/integration/roles/test_var_precedence/tasks/main.yml
@@ -1,3 +1,7 @@
+- debug: var=extra_var
+- debug: var=vars_var
+- debug: var=vars_files_var
+- debug: var=vars_files_var_role
 - assert:
     that:
       - 'extra_var == "extra_var"'
diff --git a/test/units/TestPlayVarsFiles.py b/test/units/TestPlayVarsFiles.py
index d1b1f9dfa22..f241936a12e 100644
--- a/test/units/TestPlayVarsFiles.py
+++ b/test/units/TestPlayVarsFiles.py
@@ -82,8 +82,8 @@ class TestMe(unittest.TestCase):
         os.remove(temp_path)
 
         # make sure the variable was loaded
-        assert 'foo' in play.vars, "vars_file was not loaded into play.vars"
-        assert play.vars['foo'] == 'bar', "foo was not set to bar in play.vars"
+        assert 'foo' in play.vars_file_vars, "vars_file was not loaded into play.vars_file_vars"
+        assert play.vars_file_vars['foo'] == 'bar', "foo was not set to bar in play.vars_file_vars"
 
     def test_vars_file_nonlist_error(self):
 
@@ -133,10 +133,10 @@ class TestMe(unittest.TestCase):
         os.remove(temp_path2)
 
         # make sure the variables were loaded
-        assert 'foo' in play.vars, "vars_file was not loaded into play.vars"
-        assert play.vars['foo'] == 'bar', "foo was not set to bar in play.vars"
-        assert 'baz' in play.vars, "vars_file2 was not loaded into play.vars"
-        assert play.vars['baz'] == 'bang', "baz was not set to bang in play.vars"
+        assert 'foo' in play.vars_file_vars, "vars_file was not loaded into play.vars_file_vars"
+        assert play.vars_file_vars['foo'] == 'bar', "foo was not set to bar in play.vars_file_vars"
+        assert 'baz' in play.vars_file_vars, "vars_file2 was not loaded into play.vars_file_vars"
+        assert play.vars_file_vars['baz'] == 'bang', "baz was not set to bang in play.vars_file_vars"
 
     def test_vars_files_first_found(self):
 
@@ -160,8 +160,8 @@ class TestMe(unittest.TestCase):
         os.remove(temp_path)
 
         # make sure the variable was loaded
-        assert 'foo' in play.vars, "vars_file was not loaded into play.vars"
-        assert play.vars['foo'] == 'bar', "foo was not set to bar in play.vars"
+        assert 'foo' in play.vars_file_vars, "vars_file was not loaded into play.vars_file_vars"
+        assert play.vars_file_vars['foo'] == 'bar', "foo was not set to bar in play.vars_file_vars"
 
     def test_vars_files_multiple_found(self):
 
@@ -187,9 +187,9 @@ class TestMe(unittest.TestCase):
         os.remove(temp_path2)
 
         # make sure the variables were loaded
-        assert 'foo' in play.vars, "vars_file was not loaded into play.vars"
-        assert play.vars['foo'] == 'bar', "foo was not set to bar in play.vars"
-        assert 'baz' not in play.vars, "vars_file2 was loaded after vars_file1 was loaded"
+        assert 'foo' in play.vars_file_vars, "vars_file was not loaded into play.vars_file_vars"
+        assert play.vars_file_vars['foo'] == 'bar', "foo was not set to bar in play.vars_file_vars"
+        assert 'baz' not in play.vars_file_vars, "vars_file2 was loaded after vars_file1 was loaded"
 
     def test_vars_files_assert_all_found(self):
 
@@ -227,7 +227,7 @@ class TestMe(unittest.TestCase):
     # VARIABLE PRECEDENCE TESTS
     ########################################
 
-    # On the first run vars_files are loaded into play.vars by host == None
+    # On the first run vars_files are loaded into play.vars_file_vars by host == None
     #   * only files with vars from host==None will work here
     # On the secondary run(s), a host is given and the vars_files are loaded into VARS_CACHE
     #   * this only occurs if host is not None, filename2 has vars in the name, and filename3 does not
@@ -273,8 +273,8 @@ class TestMe(unittest.TestCase):
 
     def test_vars_files_two_vars_in_name(self):
 
-        # self.vars = ds['vars']
-        # self.vars += _get_vars() ... aka extra_vars
+        # self.vars_file_vars = ds['vars']
+        # self.vars_file_vars += _get_vars() ... aka extra_vars
 
         # make a temp dir
         temp_dir = mkdtemp()
@@ -299,7 +299,7 @@ class TestMe(unittest.TestCase):
         # cleanup
         shutil.rmtree(temp_dir)
 
-        assert 'foo' in play.vars, "double var templated vars_files filename not loaded"
+        assert 'foo' in play.vars_file_vars, "double var templated vars_files filename not loaded"
     
     def test_vars_files_two_vars_different_scope(self):
 
@@ -337,7 +337,7 @@ class TestMe(unittest.TestCase):
         # cleanup
         shutil.rmtree(temp_dir)
 
-        assert 'foo' not in play.vars, \
+        assert 'foo' not in play.vars_file_vars, \
             "mixed scope vars_file loaded into play vars"
         assert 'foo' in play.playbook.VARS_CACHE['localhost'], \
             "differently scoped templated vars_files filename not loaded"
@@ -376,7 +376,7 @@ class TestMe(unittest.TestCase):
         # cleanup
         shutil.rmtree(temp_dir)
 
-        assert 'foo' not in play.vars, \
+        assert 'foo' not in play.vars_file_vars, \
             "mixed scope vars_file loaded into play vars"
         assert 'foo' in play.playbook.VARS_CACHE['localhost'], \
             "differently scoped templated vars_files filename not loaded"