diff --git a/test/v2/parsing/test_mod_args.py b/test/v2/parsing/test_mod_args.py
index f55ac29aaf..2e98cd5b00 100644
--- a/test/v2/parsing/test_mod_args.py
+++ b/test/v2/parsing/test_mod_args.py
@@ -13,67 +13,69 @@ class TestModArgsDwim(unittest.TestCase):
         pass
 
     def test_action_to_shell(self):
-        mod, args, to = self.m.parse('action', 'shell echo hi')
-        assert mod == 'shell'
+        mod, args, to = self.m.parse(dict(action='shell echo hi'))
+        assert mod == 'command'
         assert args == dict(
-            free_form = 'echo hi',
-            use_shell = True
+            _raw_params = 'echo hi',
+            _uses_shell = True,
         )
         assert to is None
 
     def test_basic_shell(self):
-        mod, args, to = self.m.parse('shell', 'echo hi')
-        assert mod == 'shell'
+        mod, args, to = self.m.parse(dict(shell='echo hi'))
+        assert mod == 'command'
         assert args == dict(
-           free_form = 'echo hi',
-           use_shell = True
+           _raw_params = 'echo hi',
+           _uses_shell = True,
         )
         assert to is None
 
     def test_basic_command(self):
-        mod, args, to = self.m.parse('command', 'echo hi')
+        mod, args, to = self.m.parse(dict(command='echo hi'))
         assert mod == 'command'
         assert args == dict(
-           free_form = 'echo hi',
-           use_shell = False
+           _raw_params = 'echo hi',
         )
         assert to is None
 
     def test_shell_with_modifiers(self):
-        mod, args, to = self.m.parse('shell', '/bin/foo creates=/tmp/baz removes=/tmp/bleep')
-        assert mod == 'shell'
+        mod, args, to = self.m.parse(dict(shell='/bin/foo creates=/tmp/baz removes=/tmp/bleep'))
+        assert mod == 'command'
         assert args == dict(
-           free_form = 'echo hi',
-           use_shell = False,
-           creates   = '/tmp/baz',
-           removes   = '/tmp/bleep'
+           creates     = '/tmp/baz',
+           removes     = '/tmp/bleep',
+           _raw_params = '/bin/foo',
+           _uses_shell = True,
         )
         assert to is None
 
     def test_normal_usage(self):
-        mod, args, to = self.m.parse('copy', 'src=a dest=b')
+        mod, args, to = self.m.parse(dict(copy='src=a dest=b'))
         assert mod == 'copy'
         assert args == dict(src='a', dest='b')
         assert to is None
 
     def test_complex_args(self):
-        mod, args, to = self.m.parse('copy', dict(src=a, dest=b))
+        mod, args, to = self.m.parse(dict(copy=dict(src='a', dest='b')))
         assert mod == 'copy'
-        assert args == dict(src = 'a', dest = 'b')
+        assert args == dict(src='a', dest='b')
         assert to is None
 
     def test_action_with_complex(self):
-        mod, args, to = self.m.parse('action', dict(module='copy',src='a',dest='b'))
-        assert mod == 'action'
-        assert args == dict(src = 'a', dest = 'b')
+        mod, args, to = self.m.parse(dict(action=dict(module='copy', src='a', dest='b')))
+        assert mod == 'copy'
+        assert args == dict(src='a', dest='b')
+        assert to is None
+
+    def test_action_with_complex_and_complex_args(self):
+        mod, args, to = self.m.parse(dict(action=dict(module='copy', args=dict(src='a', dest='b'))))
+        assert mod == 'copy'
+        assert args == dict(src='a', dest='b')
         assert to is None
 
     def test_local_action_string(self):
-        mod, args, to = self.m.parse('local_action', 'copy src=a dest=b')
+        mod, args, to = self.m.parse(dict(local_action='copy src=a dest=b'))
         assert mod == 'copy'
-        assert args == dict(src=a, dest=b)
+        assert args == dict(src='a', dest='b')
         assert to is 'localhost' 
 
-
-    
-   
diff --git a/test/v2/playbook/test_task.py b/test/v2/playbook/test_task.py
index 1404f87639..ed85ad64e7 100644
--- a/test/v2/playbook/test_task.py
+++ b/test/v2/playbook/test_task.py
@@ -36,13 +36,14 @@ class TestTask(unittest.TestCase):
         t = Task.load(basic_shell_task)
         assert t is not None
         assert t.name == basic_shell_task['name']
-        assert t.action == 'shell'
-        assert t.args == 'echo hi'
+        assert t.action == 'command'
+        assert t.args == dict(_raw_params='echo hi', _uses_shell=True)
 
     def test_load_task_kv_form(self):
         t = Task.load(kv_shell_task)
-        assert t.action == 'shell'
-        #assert t.args == 'echo hi'
+        print "task action is %s" % t.action
+        assert t.action == 'command'
+        assert t.args == dict(_raw_params='echo hi', _uses_shell=True)
 
     def test_task_auto_name(self):
         assert 'name' not in kv_shell_task
diff --git a/v2/ansible/parsing/mod_args.py b/v2/ansible/parsing/mod_args.py
index e125459037..b8f63123f9 100644
--- a/v2/ansible/parsing/mod_args.py
+++ b/v2/ansible/parsing/mod_args.py
@@ -17,6 +17,10 @@
 
 import exceptions
 
+from ansible.errors import AnsibleError
+from ansible.plugins import module_finder
+from ansible.parsing.splitter import parse_kv
+
 class ModuleArgsParser(object):
 
     """
@@ -24,25 +28,25 @@ class ModuleArgsParser(object):
 
     # legacy form (for a shell command)
     - action: shell echo hi
-   
+
     # common shorthand for local actions vs delegate_to
     - local_action: shell echo hi
 
     # most commonly:
     - copy: src=a dest=b
-      
+
     # legacy form
     - action: copy src=a dest=b
 
     # complex args form, for passing structured data
-    - copy: 
+    - copy:
         src: a
         dest: b
 
     # gross, but technically legal
     - action:
         module: copy
-        args: 
+        args:
           src: a
           dest: b
 
@@ -52,9 +56,257 @@ class ModuleArgsParser(object):
     """
 
     def __init__(self):
-        pass
+        self._ds = None
 
-    def parse(self, thing1, thing2):
-        raise exceptions.NotImplementedError
+    def _get_delegate_to(self):
+        '''
+        Returns the value of the delegate_to key from the task datastructure,
+        or None if the value was not directly specified
+        '''
+        return self._ds.get('delegate_to')
 
+    def _get_old_style_action(self):
+        '''
+        Searches the datastructure for 'action:' or 'local_action:' keywords.
+        When local_action is found, the delegate_to value is set to the localhost
+        IP, otherwise delegate_to is left as None.
+
+        Inputs:
+        - None
+
+        Outputs:
+        - None (if neither keyword is found), or a dictionary containing:
+          action:
+            the module name to be executed
+          args:
+            a dictionary containing the arguments to the module
+          delegate_to:
+            None or 'localhost'
+        '''
+
+        # determine if this is an 'action' or 'local_action'
+        if 'action' in self._ds:
+            action_data = self._ds.get('action', '')
+            delegate_to = None
+        elif 'local_action' in self._ds:
+            action_data = self._ds.get('local_action', '')
+            delegate_to = 'localhost'
+        else:
+            return None
+
+        # now we get the arguments for the module, which may be a
+        # string of key=value pairs, a dictionary of values, or a
+        # dictionary with a special 'args:' value in it
+        if isinstance(action_data, dict):
+            action = self._get_specified_module(action_data)
+            args = dict()
+            if 'args' in action_data:
+                args = self._get_args_from_ds(action, action_data)
+                del action_data['args']
+            other_args = action_data.copy()
+            # remove things we don't want in the args
+            if 'module' in other_args:
+                del other_args['module']
+            args.update(other_args)
+        elif isinstance(action_data, basestring):
+            action_data = action_data.strip()
+            if not action_data:
+                # TODO: change to an AnsibleParsingError so that the
+                #       filename/line number can be reported in the error
+                raise AnsibleError("when using 'action:' or 'local_action:', the module name must be specified")
+            else:
+                # split up the string based on spaces, where the first
+                # item specified must be a valid module name
+                parts = action_data.split(' ', 1)
+                action = parts[0]
+                if action not in module_finder:
+                    # TODO: change to an AnsibleParsingError so that the
+                    #       filename/line number can be reported in the error
+                    raise AnsibleError("the module '%s' was not found in the list of loaded modules")
+                if len(parts) > 1:
+                    args = self._get_args_from_action(action, ' '.join(parts[1:]))
+                else:
+                    args = {}
+        else:
+            # TODO: change to an AnsibleParsingError so that the
+            #       filename/line number can be reported in the error
+            raise AnsibleError('module args must be specified as a dictionary or string')
+
+        return dict(action=action, args=args, delegate_to=delegate_to)
+
+    def _get_new_style_action(self):
+        '''
+        Searches the datastructure for 'module_name:', where the module_name is a
+        valid module loaded by the module_finder plugin.
+
+        Inputs:
+        - None
+
+        Outputs:
+        - None (if no valid module is found), or a dictionary containing:
+          action:
+            the module name to be executed
+          args:
+            a dictionary containing the arguments to the module
+          delegate_to:
+            None
+        '''
+
+        # for all keys in the datastructure, check to see if the value
+        # corresponds to a module found by the module_finder plugin
+        action = None
+        for item in self._ds:
+            if item in module_finder:
+                action = item
+                break
+        else:
+            # none of the keys matched a known module name
+            return None
+
+        # now we get the arguments for the module, which may be a
+        # string of key=value pairs, a dictionary of values, or a
+        # dictionary with a special 'args:' value in it
+        action_data = self._ds.get(action, '')
+        if isinstance(action_data, dict):
+            args = dict()
+            if 'args' in action_data:
+                args = self._get_args_from_ds(action, action_data)
+                del action_data['args']
+            other_args = action_data.copy()
+            # remove things we don't want in the args
+            if 'module' in other_args:
+                del other_args['module']
+            args.update(other_args)
+        else:
+            args = self._get_args_from_action(action, action_data.strip())
+
+        return dict(action=action, args=args, delegate_to=None)
+
+    def _get_args_from_ds(self, action, action_data):
+        '''
+        Gets the module arguments from the 'args' value of the
+        action_data, when action_data is a dict. The value of
+        'args' can be either a string or a dictionary itself, so
+        we use parse_kv() to split up the key=value pairs when
+        a string is found.
+
+        Inputs:
+        - action_data:
+            a dictionary of values, which may or may not contain a
+            key named 'args'
+
+        Outputs:
+        - a dictionary of values, representing the arguments to the
+          module action specified
+        '''
+        args = action_data.get('args', {}).copy()
+        if isinstance(args, basestring):
+            if action in ('command', 'shell'):
+                args = parse_kv(args, check_raw=True)
+            else:
+                args = parse_kv(args)
+        return args
+
+    def _get_args_from_action(self, action, action_data):
+        '''
+        Gets the module arguments from the action data when it is
+        specified as a string of key=value pairs. Special handling
+        is used for the command/shell modules, which allow free-
+        form syntax for the options.
+
+        Inputs:
+        - action:
+            the module to be executed
+        - action_data:
+            a string of key=value pairs (and possibly free-form arguments)
+
+        Outputs:
+        - A dictionary of values, representing the arguments to the
+          module action specified OR a string of key=value pairs (when
+          the module action is command or shell)
+        '''
+        tokens = action_data.split()
+        if len(tokens) == 0:
+            return {}
+        else:
+            joined = " ".join(tokens)
+            if action in ('command', 'shell'):
+                return parse_kv(joined, check_raw=True)
+            else:
+                return parse_kv(joined)
+
+    def _get_specified_module(self, action_data):
+        '''
+        gets the module if specified directly in the arguments, ie:
+        - action:
+            module: foo
+
+        Inputs:
+        - action_data:
+            a dictionary of values, which may or may not contain the
+            key 'module'
+
+        Outputs:
+        - a string representing the module specified in the data, or
+          None if that key was not found
+        '''
+        return action_data.get('module')
+
+    def parse(self, ds):
+        '''
+        Given a task in one of the supported forms, parses and returns
+        returns the action, arguments, and delegate_to values for the
+        task.
+
+        Inputs:
+        - ds:
+            a dictionary datastructure representing the task as parsed
+            from a YAML file
+
+        Outputs:
+        - A tuple containing 3 values:
+          action:
+            the action (module name) to be executed
+          args:
+            the args for the action
+          delegate_to:
+            the delegate_to option (which may be None, if no delegate_to
+            option was specified and this is not a local_action)
+        '''
+
+        assert type(ds) == dict
+
+        self._ds = ds
+ 
+        # first we try to get the module action/args based on the
+        # new-style format, where the module name is the key
+        result = self._get_new_style_action()
+        if result is None:
+            # failing that, we resort to checking for the old-style syntax,
+            # where 'action' or 'local_action' is the key
+            result = self._get_old_style_action()
+            if result is None:
+                # TODO: change to an AnsibleParsingError so that the
+                #       filename/line number can be reported in the error
+                raise AnsibleError('no action specified for this task')
+
+        # if the action is set to 'shell', we switch that to 'command' and
+        # set the special parameter '_uses_shell' to true in the args dict
+        if result['action'] == 'shell':
+            result['action'] = 'command'
+            result['args']['_uses_shell'] = True
+
+        # finally, we check to see if a delegate_to value was specified
+        # in the task datastructure (and raise an error for local_action,
+        # which essentially means we're delegating to localhost)
+        specified_delegate_to = self._get_delegate_to()
+        if specified_delegate_to is not None:
+            if result['delegate_to'] is not None:
+                # TODO: change to an AnsibleParsingError so that the
+                #       filename/line number can be reported in the error
+                raise AnsibleError('delegate_to cannot be used with local_action')
+            else:
+               result['delegate_to'] = specified_delegate_to
+
+        return (result['action'], result['args'], result['delegate_to'])
 
diff --git a/v2/ansible/parsing/splitter.py b/v2/ansible/parsing/splitter.py
index 430f4e299a..48367ca5d9 100644
--- a/v2/ansible/parsing/splitter.py
+++ b/v2/ansible/parsing/splitter.py
@@ -15,8 +15,14 @@
 # You should have received a copy of the GNU General Public License
 # along with Ansible.  If not, see <http://www.gnu.org/licenses/>.
 
-def parse_kv(args):
-    ''' convert a string of key/value items to a dict '''
+def parse_kv(args, check_raw=False):
+    '''
+    Convert a string of key/value items to a dict. If any free-form params
+    are found and the check_raw option is set to True, they will be added
+    to a new parameter called '_raw_params'. If check_raw is not enabled,
+    they will simply be ignored.
+    '''
+
     options = {}
     if args is not None:
         try:
@@ -26,10 +32,31 @@ def parse_kv(args):
                 raise errors.AnsibleError("error parsing argument string, try quoting the entire line.")
             else:
                 raise
+
+        raw_params = []
         for x in vargs:
             if "=" in x:
-                k, v = x.split("=",1)
-                options[k.strip()] = unquote(v.strip())
+                k, v = x.split("=", 1)
+
+                # only internal variables can start with an underscore, so
+                # we don't allow users to set them directy in arguments
+                if k.startswith('_'):
+                    raise AnsibleError("invalid parameter specified: '%s'" % k)
+
+                # FIXME: make the retrieval of this list of shell/command
+                #        options a function, so the list is centralized
+                if check_raw and k not in ('creates', 'removes', 'chdir', 'executable', 'warn'):
+                    raw_params.append(x)
+                else:
+                    options[k.strip()] = unquote(v.strip())
+            else:
+                raw_params.append(x)
+
+        # recombine the free-form params, if any were found, and assign
+        # them to a special option for use later by the shell/command module
+        if len(raw_params) > 0:
+            options['_raw_params'] = ' '.join(raw_params)
+                
     return options
 
 def _get_quote_state(token, quote_char):
diff --git a/v2/ansible/playbook/task.py b/v2/ansible/playbook/task.py
index 856246c327..0bccedaa21 100644
--- a/v2/ansible/playbook/task.py
+++ b/v2/ansible/playbook/task.py
@@ -18,13 +18,10 @@
 from ansible.playbook.base import Base
 from ansible.playbook.attribute import Attribute, FieldAttribute
 
-# from ansible.playbook.conditional import Conditional
 from ansible.errors import AnsibleError
 
-# TODO: it would be fantastic (if possible) if a task new where in the YAML it was defined for describing
-# it in error conditions
-
 from ansible.parsing.splitter import parse_kv
+from ansible.parsing.mod_args import ModuleArgsParser
 from ansible.plugins import module_finder, lookup_finder
 
 class Task(Base):
@@ -45,7 +42,6 @@ class Task(Base):
     # validate_<attribute_name>
     # will be used if defined
     # might be possible to define others 
-
   
     _args                 = FieldAttribute(isa='dict')
     _action               = FieldAttribute(isa='string')
@@ -60,9 +56,6 @@ class Task(Base):
     _first_available_file = FieldAttribute(isa='list')
     _ignore_errors        = FieldAttribute(isa='bool')
 
-    # FIXME: this should not be a Task
-    # include             = FieldAttribute(isa='string')
-
     _loop                 = FieldAttribute(isa='string', private=True)
     _loop_args            = FieldAttribute(isa='list', private=True)
     _local_action         = FieldAttribute(isa='string')
@@ -102,16 +95,19 @@ class Task(Base):
        elif self.name:
            return self.name
        else:
-           return "%s %s" % (self.action, self._merge_kv(self.args))
+           flattened_args = self._merge_kv(self.args)
+           return "%s %s" % (self.action, flattened_args)
 
     def _merge_kv(self, ds):
         if ds is None:
             return ""
         elif isinstance(ds, basestring):
             return ds
-        elif instance(ds, dict):
+        elif isinstance(ds, dict):
             buf = ""
             for (k,v) in ds.iteritems():
+                if k.startswith('_'):
+                    continue
                 buf = buf + "%s=%s " % (k,v)
             buf = buf.strip()
             return buf
@@ -125,27 +121,6 @@ class Task(Base):
         ''' returns a human readable representation of the task '''
         return "TASK: %s" % self.get_name()
                 
-    def _parse_old_school_action(self, v):
-        ''' given a action/local_action line, return the module and args ''' 
-        tokens = v.split()
-        if len(tokens) < 2:
-            return [v,{}]
-        else:
-            if v not in [ 'command', 'shell' ]:
-                joined = " ".join(tokens[1:])
-                return [tokens[0], parse_kv(joined)]
-            else:
-                return [tokens[0], joined] 
-
-    def _munge_action(self, ds, new_ds, k, v):
-        ''' take a module name and split into action and args '''
-
-        if self._action.value is not None or 'action' in ds or 'local_action' in ds:
-            raise AnsibleError("duplicate action in task: %s" % k)
-        new_ds['action'] = k
-        new_ds['args'] = v
-                
-
     def _munge_loop(self, ds, new_ds, k, v):
         ''' take a lookup plugin name and store it correctly '''
 
@@ -154,22 +129,6 @@ class Task(Base):
         new_ds['loop'] = k
         new_ds['loop_args'] = v
                 
-    def _munge_action2(self, ds, new_ds, k, v, local=False):
-        ''' take an old school action/local_action and reformat it '''
-
-        if isinstance(v, basestring):
-            tokens = self._parse_old_school_action(v)
-            new_ds['action'] = tokens[0]
-            if 'args' in ds:
-                raise AnsibleError("unexpected and redundant 'args'")
-                new_ds['args'] = args
-                if local:
-                    if 'delegate_to' in ds:
-                       raise AnsbileError("local_action and action conflict")
-                    new_ds['delegate_to'] = 'localhost'
-        else:
-            raise AnsibleError("unexpected use of 'action'")
-
     def munge(self, ds):
         ''' 
         tasks are especially complex arguments so need pre-processing.
@@ -178,18 +137,31 @@ class Task(Base):
 
         assert isinstance(ds, dict)
 
+        # the new, cleaned datastructure, which will have legacy
+        # items reduced to a standard structure suitable for the
+        # attributes of the task class
         new_ds = dict()
+
+        # use the args parsing class to determine the action, args,
+        # and the delegate_to value from the various possible forms
+        # supported as legacy
+        args_parser = ModuleArgsParser()
+        (action, args, delegate_to) = args_parser.parse(ds)
+
+        new_ds['action']      = action
+        new_ds['args']        = args
+        new_ds['delegate_to'] = delegate_to
+
         for (k,v) in ds.iteritems():
-            if k in module_finder:
-                self._munge_action(ds, new_ds, k, v)
+            if k in ('action', 'local_action', 'args', 'delegate_to') or k == action:
+                # we don't want to re-assign these values, which were
+                # determined by the ModuleArgsParser() above
+                continue
             elif "with_%s" % k in lookup_finder:
                 self._munge_loop(ds, new_ds, k, v)
-            elif k == 'action':
-                self._munge_action2(ds, new_ds, k, v) 
-            elif k == 'local_action':
-                self._munge_action2(ds, new_ds, k, v, local=True)
             else:
                 new_ds[k] = v
+
         return new_ds