diff --git a/lib/ClusterShell/Gateway.py b/lib/ClusterShell/Gateway.py
index 319d07ef..ec3c1db4 100755
--- a/lib/ClusterShell/Gateway.py
+++ b/lib/ClusterShell/Gateway.py
@@ -255,6 +255,11 @@ def recv_ctl(self, msg):
                 task._info.update(taskinfo)
                 task.set_info('print_debug', _gw_print_debug)
 
+                for infokey in taskinfo:
+                    if infokey.startswith('tree_default:'):
+                        self.logger.debug('Setting default %s to %s', infokey[13:], taskinfo[infokey])
+                        task.set_default(infokey[13:], taskinfo[infokey])
+
                 if task.info('debug'):
                     self.logger.setLevel(logging.DEBUG)
 
diff --git a/lib/ClusterShell/Task.py b/lib/ClusterShell/Task.py
index 77aeb2be..b9142e6d 100644
--- a/lib/ClusterShell/Task.py
+++ b/lib/ClusterShell/Task.py
@@ -58,7 +58,7 @@
     basestring = str
 
 from ClusterShell.Defaults import config_paths, DEFAULTS
-from ClusterShell.Defaults import _local_workerclass, _distant_workerclass
+from ClusterShell.Defaults import _local_workerclass, _distant_workerclass, _load_workerclass
 from ClusterShell.Engine.Engine import EngineAbortException
 from ClusterShell.Engine.Engine import EngineTimeoutException
 from ClusterShell.Engine.Engine import EngineAlreadyRunningError
@@ -470,6 +470,10 @@ def set_default(self, default_key, value):
         self._default_lock.acquire()
         try:
             self._default[default_key] = value
+            if default_key == 'local_workername':
+                self._default['local_worker'] = _load_workerclass(value)
+            elif default_key == 'distant_workername':
+                self._default['distant_worker'] = _load_workerclass(value)
         finally:
             self._default_lock.release()
 
@@ -510,6 +514,8 @@ def set_info(self, info_key, value):
           - "command_timeout": Time in seconds to wait for a command to
             complete before aborting (default: 0, which means
             unlimited).
+          - "tree_default:<key>": In tree mode, overrides the key <key>
+            in Defaults (settings normally set in defaults.conf)
 
         Threading considerations
         ========================
diff --git a/lib/ClusterShell/Worker/Tree.py b/lib/ClusterShell/Worker/Tree.py
index 7f9aa88f..8dad9289 100644
--- a/lib/ClusterShell/Worker/Tree.py
+++ b/lib/ClusterShell/Worker/Tree.py
@@ -280,11 +280,12 @@ def _launch(self, nodes):
                                                  tree=False)
                 else:
                     assert self.source is None
-                    worker = ExecWorker(nodes=targets,
-                                        command=self.command,
-                                        handler=self.metahandler,
-                                        timeout=self.timeout,
-                                        stderr=self.stderr)
+                    workerclass = self.task.default('local_worker')
+                    worker = workerclass(nodes=targets,
+                                         command=self.command,
+                                         handler=self.metahandler,
+                                         timeout=self.timeout,
+                                         stderr=self.stderr)
                     self.task.schedule(worker)
 
                 self.workers.append(worker)
diff --git a/tests/TreeWorkerTest.py b/tests/TreeWorkerTest.py
index ba12ccfe..d5c221a4 100644
--- a/tests/TreeWorkerTest.py
+++ b/tests/TreeWorkerTest.py
@@ -191,6 +191,22 @@ def test_tree_run_noremote(self):
         self.assertEqual(teh.ev_close_cnt, 1)
         self.assertEqual(teh.last_read, NODE_DISTANT.encode('ascii'))
 
+    def test_tree_run_noremote_alt_localworker(self):
+        """test tree run with remote=False and a non-exec localworker"""
+        teh = TEventHandler()
+        self.task.set_info('tree_default:local_workername', 'ssh')
+        self.task.run('echo %h', nodes=NODE_DISTANT, handler=teh, remote=False)
+        self.assertEqual(teh.ev_start_cnt, 1)
+        self.assertEqual(teh.ev_pickup_cnt, 1)
+        self.assertEqual(teh.ev_read_cnt, 1)
+        self.assertEqual(teh.ev_written_cnt, 0)
+        self.assertEqual(teh.ev_hup_cnt, 1)
+        self.assertEqual(teh.ev_timedout_cnt, 0)
+        self.assertEqual(teh.ev_close_cnt, 1)
+        # The exec worker will expand %h to the host, but ssh will just echo '%h'
+        self.assertEqual(teh.last_read, '%h'.encode('ascii'))
+        del self.task._info['tree_default:local_workername']
+
     def test_tree_run_direct(self):
         """test tree run with direct target, in topology"""
         teh = TEventHandler()