diff --git a/src/python/pants/backend/jvm/tasks/nailgun_task.py b/src/python/pants/backend/jvm/tasks/nailgun_task.py
index 7d0fd4a134b..8160b8489fc 100644
--- a/src/python/pants/backend/jvm/tasks/nailgun_task.py
+++ b/src/python/pants/backend/jvm/tasks/nailgun_task.py
@@ -9,11 +9,11 @@
 
 from pants.backend.jvm.tasks.jvm_tool_task_mixin import JvmToolTaskMixin
 from pants.base.exceptions import TaskError
+from pants.init.subprocess import Subprocess
 from pants.java import util
 from pants.java.executor import SubprocessExecutor
 from pants.java.jar.jar_dependency import JarDependency
 from pants.java.nailgun_executor import NailgunExecutor, NailgunProcessGroup
-from pants.pantsd.subsystem.subprocess import Subprocess
 from pants.task.task import Task, TaskBase
 
 
diff --git a/src/python/pants/bin/BUILD b/src/python/pants/bin/BUILD
index 3aba4107c20..2cba8ac8670 100644
--- a/src/python/pants/bin/BUILD
+++ b/src/python/pants/bin/BUILD
@@ -34,10 +34,12 @@ python_library(
     'src/python/pants/init',
     'src/python/pants/option',
     'src/python/pants/reporting',
+    'src/python/pants/pantsd:pants_daemon_launcher',
     'src/python/pants/scm/subsystems:changed',
     'src/python/pants/subsystem',
     'src/python/pants/task',
     'src/python/pants/util:contextutil',
+    'src/python/pants/util:collections',
     'src/python/pants/util:dirutil',
     'src/python/pants/util:filtering',
     'src/python/pants/util:memo',
diff --git a/src/python/pants/bin/goal_runner.py b/src/python/pants/bin/goal_runner.py
index 09eb28f8922..b7c36deee44 100644
--- a/src/python/pants/bin/goal_runner.py
+++ b/src/python/pants/bin/goal_runner.py
@@ -23,7 +23,7 @@
 from pants.goal.goal import Goal
 from pants.goal.run_tracker import RunTracker
 from pants.help.help_printer import HelpPrinter
-from pants.init.pants_daemon_launcher import PantsDaemonLauncher
+from pants.init.subprocess import Subprocess
 from pants.init.target_roots import TargetRoots
 from pants.java.nailgun_executor import NailgunProcessGroup
 from pants.reporting.reporting import Reporting
@@ -152,14 +152,7 @@ def generate_targets(specs):
 
     return list(generate_targets(specs))
 
-  def _maybe_launch_pantsd(self, pantsd_launcher):
-    """Launches pantsd if configured to do so."""
-    if self._global_options.enable_pantsd:
-      # Avoid runtracker output if pantsd is disabled. Otherwise, show up to inform the user its on.
-      with self._run_tracker.new_workunit(name='pantsd', labels=[WorkUnitLabel.SETUP]):
-        pantsd_launcher.maybe_launch()
-
-  def _setup_context(self, pantsd_launcher):
+  def _setup_context(self):
     with self._run_tracker.new_workunit(name='setup', labels=[WorkUnitLabel.SETUP]):
       self._build_graph, self._address_mapper, spec_roots = self._init_graph(
         self._global_options.enable_v2_engine,
@@ -189,15 +182,12 @@ def _setup_context(self, pantsd_launcher):
                         build_graph=self._build_graph,
                         build_file_parser=self._build_file_parser,
                         address_mapper=self._address_mapper,
-                        invalidation_report=invalidation_report,
-                        pantsd_launcher=pantsd_launcher)
+                        invalidation_report=invalidation_report)
       return goals, context
 
   def setup(self):
-    pantsd_launcher = PantsDaemonLauncher.Factory.global_instance().create(EngineInitializer)
-    self._maybe_launch_pantsd(pantsd_launcher)
     self._handle_help(self._help_request)
-    goals, context = self._setup_context(pantsd_launcher)
+    goals, context = self._setup_context()
     return GoalRunner(context=context,
                       goals=goals,
                       run_tracker=self._run_tracker,
@@ -234,7 +224,7 @@ def subsystems(cls):
       RunTracker,
       Changed.Factory,
       BinaryUtil.Factory,
-      PantsDaemonLauncher.Factory,
+      Subprocess.Factory
     }
 
   def _execute_engine(self):
diff --git a/src/python/pants/bin/pants_runner.py b/src/python/pants/bin/pants_runner.py
index 83f09ac171f..895d371b19c 100644
--- a/src/python/pants/bin/pants_runner.py
+++ b/src/python/pants/bin/pants_runner.py
@@ -29,29 +29,24 @@ def __init__(self, exiter, args=None, env=None):
     self._args = args or sys.argv
     self._env = env or os.environ
 
-  def _run(self, is_remote, exiter, args, env, process_metadata_dir=None, options_bootstrapper=None):
-    if is_remote:
+  def run(self):
+    options_bootstrapper = OptionsBootstrapper(env=self._env, args=self._args)
+    bootstrap_options = options_bootstrapper.get_bootstrap_options().for_global_scope()
+
+    if bootstrap_options.enable_pantsd:
       try:
-        return RemotePantsRunner(exiter, args, env, process_metadata_dir).run()
-      except RemotePantsRunner.RECOVERABLE_EXCEPTIONS as e:
-        # N.B. RemotePantsRunner will raise one of RECOVERABLE_EXCEPTIONS in the event we
-        # encounter a failure while discovering or initially connecting to the pailgun. In
-        # this case, we fall back to LocalPantsRunner which seamlessly executes the requested
-        # run and bootstraps pantsd for use in subsequent runs.
-        logger.debug('caught client exception: {!r}, falling back to LocalPantsRunner'.format(e))
+        return RemotePantsRunner(self._exiter,
+                                 self._args,
+                                 self._env,
+                                 bootstrap_options.pants_subprocessdir,
+                                 bootstrap_options).run()
+      except RemotePantsRunner.Fallback as e:
+        logger.debug('caught client exception: {!r}, falling back to non-daemon mode'.format(e))
 
     # N.B. Inlining this import speeds up the python thin client run by about 100ms.
     from pants.bin.local_pants_runner import LocalPantsRunner
 
-    return LocalPantsRunner(exiter, args, env, options_bootstrapper=options_bootstrapper).run()
-
-  def run(self):
-    options_bootstrapper = OptionsBootstrapper(env=self._env, args=self._args)
-    global_bootstrap_options = options_bootstrapper.get_bootstrap_options().for_global_scope()
-
-    return self._run(is_remote=global_bootstrap_options.enable_pantsd,
-                     exiter=self._exiter,
-                     args=self._args,
-                     env=self._env,
-                     process_metadata_dir=global_bootstrap_options.pants_subprocessdir,
-                     options_bootstrapper=options_bootstrapper)
+    return LocalPantsRunner(self._exiter,
+                            self._args,
+                            self._env,
+                            options_bootstrapper=options_bootstrapper).run()
diff --git a/src/python/pants/bin/remote_pants_runner.py b/src/python/pants/bin/remote_pants_runner.py
index f0c9c7816eb..47e07fe99b0 100644
--- a/src/python/pants/bin/remote_pants_runner.py
+++ b/src/python/pants/bin/remote_pants_runner.py
@@ -5,30 +5,41 @@
 from __future__ import (absolute_import, division, generators, nested_scopes, print_function,
                         unicode_literals, with_statement)
 
+import logging
 import signal
 import sys
 from contextlib import contextmanager
 
+from pants.bin.engine_initializer import EngineInitializer
 from pants.java.nailgun_client import NailgunClient
 from pants.java.nailgun_protocol import NailgunProtocol
-from pants.pantsd.process_manager import ProcessMetadataManager
+from pants.pantsd.pants_daemon_launcher import PantsDaemonLauncher
+from pants.util.collections import combined_dict
+
+
+logger = logging.getLogger(__name__)
 
 
 class RemotePantsRunner(object):
   """A thin client variant of PantsRunner."""
 
-  class PortNotFound(Exception): pass
+  class Fallback(Exception):
+    """Raised when fallback to an alternate execution mode is requested."""
+
+  class PortNotFound(Exception):
+    """Raised when the pailgun port can't be found."""
 
   PANTS_COMMAND = 'pants'
   RECOVERABLE_EXCEPTIONS = (PortNotFound, NailgunClient.NailgunConnectionError)
 
-  def __init__(self, exiter, args, env, process_metadata_dir=None,
-               stdin=None, stdout=None, stderr=None):
+  def __init__(self, exiter, args, env, process_metadata_dir,
+               bootstrap_options, stdin=None, stdout=None, stderr=None):
     """
     :param Exiter exiter: The Exiter instance to use for this run.
     :param list args: The arguments (e.g. sys.argv) for this run.
     :param dict env: The environment (e.g. os.environ) for this run.
     :param str process_metadata_dir: The directory in which process metadata is kept.
+    :param Options bootstrap_options: The Options bag containing the bootstrap options.
     :param file stdin: The stream representing stdin.
     :param file stdout: The stream representing stdout.
     :param file stderr: The stream representing stderr.
@@ -37,17 +48,11 @@ def __init__(self, exiter, args, env, process_metadata_dir=None,
     self._args = args
     self._env = env
     self._process_metadata_dir = process_metadata_dir
+    self._bootstrap_options = bootstrap_options
     self._stdin = stdin or sys.stdin
     self._stdout = stdout or sys.stdout
     self._stderr = stderr or sys.stderr
-    self._port = self._retrieve_pailgun_port()
-    if not self._port:
-      raise self.PortNotFound('unable to locate pailgun port!')
-
-  @staticmethod
-  def _combine_dicts(*dicts):
-    """Combine one or more dicts into a new, unified dict (dicts to the right take precedence)."""
-    return {k: v for d in dicts for k, v in d.items()}
+    self._launcher = PantsDaemonLauncher(self._bootstrap_options, EngineInitializer)
 
   @contextmanager
   def _trapped_control_c(self, client):
@@ -62,17 +67,36 @@ def handle_control_c(signum, frame):
     finally:
       signal.signal(signal.SIGINT, existing_sigint_handler)
 
-  def _retrieve_pailgun_port(self):
-    return ProcessMetadataManager(
-      self._process_metadata_dir).read_metadata_by_name('pantsd', 'socket_pailgun', int)
+  def _setup_logging(self):
+    """Sets up basic stdio logging for the thin client."""
+    log_level = logging.getLevelName(self._bootstrap_options.level.upper())
 
-  def run(self, args=None):
+    formatter = logging.Formatter('%(levelname)s] %(message)s')
+    handler = logging.StreamHandler(sys.stdout)
+    handler.setLevel(log_level)
+    handler.setFormatter(formatter)
+
+    root = logging.getLogger()
+    root.setLevel(log_level)
+    root.addHandler(handler)
+
+  def _find_or_launch_pantsd(self):
+    """Launches pantsd if configured to do so.
+
+    :returns: The port pantsd can be found on.
+    :rtype: int
+    """
+    return self._launcher.maybe_launch()
+
+  def _connect_and_execute(self, port):
     # Merge the nailgun TTY capability environment variables with the passed environment dict.
     ng_env = NailgunProtocol.isatty_to_env(self._stdin, self._stdout, self._stderr)
-    modified_env = self._combine_dicts(self._env, ng_env)
+    modified_env = combined_dict(self._env, ng_env)
+
+    assert isinstance(port, int), 'port {} is not an integer!'.format(port)
 
     # Instantiate a NailgunClient.
-    client = NailgunClient(port=self._port,
+    client = NailgunClient(port=port,
                            ins=self._stdin,
                            out=self._stdout,
                            err=self._stderr,
@@ -84,3 +108,13 @@ def run(self, args=None):
 
     # Exit.
     self._exiter.exit(result)
+
+  def run(self, args=None):
+    self._setup_logging()
+    port = self._find_or_launch_pantsd()
+
+    logger.debug('connecting to pailgun on port {}'.format(port))
+    try:
+      self._connect_and_execute(port)
+    except self.RECOVERABLE_EXCEPTIONS as e:
+      raise self.Fallback(e)
diff --git a/src/python/pants/core_tasks/pantsd_kill.py b/src/python/pants/core_tasks/pantsd_kill.py
index 187a803b903..bccf2dd4c1f 100644
--- a/src/python/pants/core_tasks/pantsd_kill.py
+++ b/src/python/pants/core_tasks/pantsd_kill.py
@@ -6,6 +6,7 @@
                         unicode_literals, with_statement)
 
 from pants.base.exceptions import TaskError
+from pants.pantsd.pants_daemon_launcher import PantsDaemonLauncher
 from pants.pantsd.process_manager import ProcessManager
 from pants.task.task import Task
 
@@ -15,6 +16,6 @@ class PantsDaemonKill(Task):
 
   def execute(self):
     try:
-      self.context.pantsd_launcher.terminate()
+      PantsDaemonLauncher(self.get_options()).terminate()
     except ProcessManager.NonResponsiveProcess as e:
       raise TaskError('failure while terminating pantsd: {}'.format(e))
diff --git a/src/python/pants/core_tasks/register.py b/src/python/pants/core_tasks/register.py
index bf8bd3dfcb7..48c68f68cf5 100644
--- a/src/python/pants/core_tasks/register.py
+++ b/src/python/pants/core_tasks/register.py
@@ -59,7 +59,9 @@ def register_goals():
   # Pantsd.
   kill_pantsd = task(name='kill-pantsd', action=PantsDaemonKill)
   kill_pantsd.install()
-  kill_pantsd.install('clean-all')
+  # Kill pantsd/watchman first, so that they're not using any files
+  # in .pants.d at the time of removal.
+  kill_pantsd.install('clean-all', first=True)
 
   # Reporting server.
   # TODO: The reporting server should be subsumed into pantsd, and not run via a task.
diff --git a/src/python/pants/engine/native.py b/src/python/pants/engine/native.py
index 9a3468d9796..afa4077f814 100644
--- a/src/python/pants/engine/native.py
+++ b/src/python/pants/engine/native.py
@@ -561,9 +561,9 @@ class Native(object):
   """Encapsulates fetching a platform specific version of the native portion of the engine."""
 
   @staticmethod
-  def create(options):
+  def create(bootstrap_options):
     """:param options: Any object that provides access to bootstrap option values."""
-    return Native(options.native_engine_visualize_to)
+    return Native(bootstrap_options.native_engine_visualize_to)
 
   def __init__(self, visualize_to_dir):
     """
diff --git a/src/python/pants/goal/context.py b/src/python/pants/goal/context.py
index 1abc11e0071..bb6143ae968 100644
--- a/src/python/pants/goal/context.py
+++ b/src/python/pants/goal/context.py
@@ -61,7 +61,7 @@ def fatal(self, *msg_elements):
   def __init__(self, options, run_tracker, target_roots,
                requested_goals=None, target_base=None, build_graph=None,
                build_file_parser=None, address_mapper=None, console_outstream=None, scm=None,
-               workspace=None, invalidation_report=None, pantsd_launcher=None):
+               workspace=None, invalidation_report=None):
     self._options = options
     self.build_graph = build_graph
     self.build_file_parser = build_file_parser
@@ -80,7 +80,6 @@ def __init__(self, options, run_tracker, target_roots,
     self._workspace = workspace or (ScmWorkspace(self._scm) if self._scm else None)
     self._replace_targets(target_roots)
     self._invalidation_report = invalidation_report
-    self._pantsd_launcher = pantsd_launcher
 
   @property
   def options(self):
@@ -151,10 +150,6 @@ def workspace(self):
   def invalidation_report(self):
     return self._invalidation_report
 
-  @property
-  def pantsd_launcher(self):
-    return self._pantsd_launcher
-
   def __str__(self):
     ident = Target.identify(self.targets())
     return 'Context(id:{}, targets:{})'.format(ident, self.targets())
diff --git a/src/python/pants/init/BUILD b/src/python/pants/init/BUILD
index 9b0bee315dd..4b11180e6c3 100644
--- a/src/python/pants/init/BUILD
+++ b/src/python/pants/init/BUILD
@@ -1,7 +1,6 @@
 # Copyright 2017 Pants project contributors (see CONTRIBUTORS.md).
 # Licensed under the Apache License, Version 2.0 (see LICENSE).
 
-
 python_library(
   dependencies=[
     ':plugins',
@@ -21,12 +20,6 @@ python_library(
     'src/python/pants/goal:run_tracker',
     'src/python/pants/logging',
     'src/python/pants/option',
-    'src/python/pants/pantsd:pants_daemon',
-    'src/python/pants/pantsd/service:fs_event_service',
-    'src/python/pants/pantsd/service:pailgun_service',
-    'src/python/pants/pantsd/service:scheduler_service',
-    'src/python/pants/pantsd/subsystem:subprocess',
-    'src/python/pants/pantsd/subsystem:watchman_launcher',
     'src/python/pants/process',
     'src/python/pants/python',
     'src/python/pants/subsystem',
diff --git a/src/python/pants/init/pants_daemon_launcher.py b/src/python/pants/init/pants_daemon_launcher.py
deleted file mode 100644
index dc675386c83..00000000000
--- a/src/python/pants/init/pants_daemon_launcher.py
+++ /dev/null
@@ -1,179 +0,0 @@
-# coding=utf-8
-# Copyright 2015 Pants project contributors (see CONTRIBUTORS.md).
-# Licensed under the Apache License, Version 2.0 (see LICENSE).
-
-from __future__ import (absolute_import, division, generators, nested_scopes, print_function,
-                        unicode_literals, with_statement)
-
-import logging
-import os
-
-from pants.base.build_environment import get_buildroot
-from pants.binaries.binary_util import BinaryUtil
-from pants.engine.native import Native
-from pants.init.target_roots import TargetRoots
-from pants.init.util import clean_global_runtime_state
-from pants.pantsd.pants_daemon import PantsDaemon
-from pants.pantsd.service.fs_event_service import FSEventService
-from pants.pantsd.service.pailgun_service import PailgunService
-from pants.pantsd.service.scheduler_service import SchedulerService
-from pants.pantsd.subsystem.subprocess import Subprocess
-from pants.pantsd.subsystem.watchman_launcher import WatchmanLauncher
-from pants.process.lock import OwnerPrintingInterProcessFileLock
-from pants.subsystem.subsystem import Subsystem
-from pants.util.memo import testable_memoized_property
-
-
-class PantsDaemonLauncher(object):
-  """A subsystem that manages the configuration and launching of pantsd."""
-
-  class Factory(Subsystem):
-    options_scope = 'pantsd'
-
-    @classmethod
-    def register_options(cls, register):
-      register('--pailgun-host', advanced=True, default='127.0.0.1',
-               help='The host to bind the pants nailgun server to.')
-      register('--pailgun-port', advanced=True, type=int, default=0,
-               help='The port to bind the pants nailgun server to. Defaults to a random port.')
-      register('--log-dir', advanced=True, default=None,
-               help='The directory to log pantsd output to.')
-      register('--fs-event-detection', advanced=True, type=bool,
-               removal_version='1.5.0.dev0',
-               removal_hint='This option is now implied by `--enable-pantsd`.',
-               help='Whether or not to use filesystem event detection.')
-      register('--fs-event-workers', advanced=True, type=int, default=4,
-               help='The number of workers to use for the filesystem event service executor pool.')
-
-    @classmethod
-    def subsystem_dependencies(cls):
-      return super(PantsDaemonLauncher.Factory,
-                   cls).subsystem_dependencies() + (WatchmanLauncher.Factory, Subprocess.Factory,
-                                                    BinaryUtil.Factory)
-
-    def create(self, engine_initializer=None):
-      """
-      :param class engine_initializer: The class representing the EngineInitializer. Only necessary
-                                       for startup.
-      """
-      build_root = get_buildroot()
-      options = self.global_instance().get_options()
-      return PantsDaemonLauncher(build_root=build_root,
-                                 engine_initializer=engine_initializer,
-                                 options=options)
-
-  def __init__(self,
-               build_root,
-               engine_initializer,
-               options):
-    """
-    :param str build_root: The path of the build root.
-    :param class engine_initializer: The class representing the EngineInitializer.
-    """
-    self._build_root = build_root
-    self._engine_initializer = engine_initializer
-
-    # The options we register directly.
-    self._pailgun_host = options.pailgun_host
-    self._pailgun_port = options.pailgun_port
-    self._log_dir = options.log_dir
-    self._fs_event_workers = options.fs_event_workers
-
-    # Values derived from global options (which our scoped options inherit).
-    self._pants_workdir = options.pants_workdir
-    self._log_level = options.level.upper()
-    self._pants_ignore_patterns = options.pants_ignore
-    self._build_ignore_patterns = options.build_ignore
-    self._exclude_target_regexp = options.exclude_target_regexp
-    self._subproject_roots = options.subproject_roots
-    # Native.create() reads global options, which, thanks to inheritance, it can
-    # read them via our scoped options.
-    self._native = Native.create(options)
-    # TODO(kwlzn): Thread filesystem path ignores here to Watchman's subscription registration.
-
-    lock_location = os.path.join(self._build_root, '.pantsd.startup')
-    self._lock = OwnerPrintingInterProcessFileLock(lock_location)
-    self._logger = logging.getLogger(__name__)
-
-  @testable_memoized_property
-  def pantsd(self):
-    return PantsDaemon(
-      self._build_root,
-      self._pants_workdir,
-      self._log_level,
-      self._native,
-      self._log_dir,
-      reset_func=clean_global_runtime_state
-    )
-
-  @testable_memoized_property
-  def watchman_launcher(self):
-    return WatchmanLauncher.Factory.global_instance().create()
-
-  def _setup_services(self, watchman):
-    """Initialize pantsd services.
-
-    :returns: A tuple of (`tuple` service_instances, `dict` port_map).
-    """
-    # N.B. This inline import is currently necessary to avoid a circular reference in the import
-    # of LocalPantsRunner for use by DaemonPantsRunner. This is because LocalPantsRunner must
-    # ultimately import the pantsd services in order to itself launch pantsd.
-    from pants.bin.daemon_pants_runner import DaemonExiter, DaemonPantsRunner
-
-    legacy_graph_helper = self._engine_initializer.setup_legacy_graph(
-      self._pants_ignore_patterns,
-      self._pants_workdir,
-      native=self._native,
-      build_ignore_patterns=self._build_ignore_patterns,
-      exclude_target_regexps=self._exclude_target_regexp,
-      subproject_roots=self._subproject_roots,
-    )
-
-    fs_event_service = FSEventService(watchman, self._build_root, self._fs_event_workers)
-    scheduler_service = SchedulerService(fs_event_service, legacy_graph_helper)
-    pailgun_service = PailgunService(
-      bind_addr=(self._pailgun_host, self._pailgun_port),
-      exiter_class=DaemonExiter,
-      runner_class=DaemonPantsRunner,
-      target_roots_class=TargetRoots,
-      scheduler_service=scheduler_service
-    )
-
-    return (
-      # Use the schedulers reentrant lock as the daemon's global lock.
-      legacy_graph_helper.scheduler.lock,
-      # Services.
-      (fs_event_service, scheduler_service, pailgun_service),
-      # Port map.
-      dict(pailgun=pailgun_service.pailgun_port)
-    )
-
-  def _launch_pantsd(self):
-    # Launch Watchman (if so configured).
-    watchman = self.watchman_launcher.maybe_launch()
-
-    # Initialize pantsd services.
-    lock, services, port_map = self._setup_services(watchman)
-
-    # Setup and fork pantsd.
-    self.pantsd.set_lock(lock)
-    self.pantsd.set_services(services)
-    self.pantsd.set_socket_map(port_map)
-    self.pantsd.daemonize()
-
-    # Wait up to 10 seconds for pantsd to write its pidfile so we can display the pid to the user.
-    self.pantsd.await_pid(10)
-
-  def maybe_launch(self):
-    self._logger.debug('acquiring lock: {}'.format(self._lock))
-    with self._lock:
-      if not self.pantsd.is_alive():
-        self._logger.debug('launching pantsd')
-        self._launch_pantsd()
-    self._logger.debug('released lock: {}'.format(self._lock))
-
-    self._logger.debug('pantsd is running at pid {}'.format(self.pantsd.pid))
-
-  def terminate(self):
-    self.pantsd.terminate()
-    self.watchman_launcher.terminate()
diff --git a/src/python/pants/pantsd/subsystem/subprocess.py b/src/python/pants/init/subprocess.py
similarity index 100%
rename from src/python/pants/pantsd/subsystem/subprocess.py
rename to src/python/pants/init/subprocess.py
diff --git a/src/python/pants/option/global_options.py b/src/python/pants/option/global_options.py
index 4bc3bdfded1..f5e9cdddca4 100644
--- a/src/python/pants/option/global_options.py
+++ b/src/python/pants/option/global_options.py
@@ -36,6 +36,9 @@ def register_bootstrap_options(cls, register):
     status as "bootstrap options" is only pertinent during option registration.
     """
     buildroot = get_buildroot()
+    default_distdir_name = 'dist'
+    default_distdir = os.path.join(buildroot, default_distdir_name)
+    default_rel_distdir = '/{}/'.format(default_distdir_name)
 
     # Although logging supports the WARN level, its not documented and could conceivably be yanked.
     # Since pants has supported 'warn' since inception, leave the 'warn' choice as-is but explicitly
@@ -89,8 +92,10 @@ def register_bootstrap_options(cls, register):
              default=os.path.join(buildroot, 'build-support'),
              help='Use support files from this dir.')
     register('--pants-distdir', advanced=True, metavar='<dir>',
-             default=os.path.join(buildroot, 'dist'),
-             help='Write end-product artifacts to this dir.')
+             default=default_distdir,
+             help='Write end-product artifacts to this dir. If you modify this path, you '
+                  'should also update --build-ignore and --pants-ignore to include the '
+                  'custom dist dir path as well.')
     register('--pants-subprocessdir', advanced=True, default=os.path.join(buildroot, '.pids'),
              help='The directory to use for tracking subprocess metadata, if any. This should '
                   'live outside of the dir used by `--pants-workdir` to allow for tracking '
@@ -115,6 +120,25 @@ def register_bootstrap_options(cls, register):
              help='Read additional specs from this file, one per line')
     register('--verify-config', type=bool, default=True,
              help='Verify that all config file values correspond to known options.')
+    register('--build-ignore', advanced=True, type=list, fromfile=True,
+             default=['.*/', default_rel_distdir, 'bower_components/',
+                      'node_modules/', '*.egg-info/'],
+             help='Paths to ignore when identifying BUILD files. '
+                  'This does not affect any other filesystem operations. '
+                  'Patterns use the gitignore pattern syntax (https://git-scm.com/docs/gitignore).')
+    register('--pants-ignore', advanced=True, type=list, fromfile=True,
+             default=['.*/', default_rel_distdir],
+             help='Paths to ignore for all filesystem operations performed by pants '
+                  '(e.g. BUILD file scanning, glob matching, etc). '
+                  'Patterns use the gitignore syntax (https://git-scm.com/docs/gitignore). '
+                  'This currently only affects the v2 engine. '
+                  'To experiment with v2 engine, try --enable-v2-engine option.')
+    register('--exclude-target-regexp', advanced=True, type=list, default=[],
+             metavar='<regexp>',
+             help='Exclude target roots that match these regexes.')
+    register('--subproject-roots', type=list, advanced=True, fromfile=True, default=[],
+             help='Paths that correspond with build roots for any subproject that this '
+                  'project depends on.')
 
     # These logging options are registered in the bootstrap phase so that plugins can log during
     # registration and not so that their values can be interpolated in configs.
@@ -158,6 +182,34 @@ def register_bootstrap_options(cls, register):
              help=('Maps output of uname for a machine to a binary search path. e.g. '
              '{("darwin", "15"): ["mac", "10.11"]), ("linux", "arm32"): ["linux", "arm32"]}'))
 
+    # Pants Daemon options.
+    register('--pantsd-pailgun-host', advanced=True, default='127.0.0.1',
+             help='The host to bind the pants nailgun server to.')
+    register('--pantsd-pailgun-port', advanced=True, type=int, default=0,
+             help='The port to bind the pants nailgun server to. Defaults to a random port.')
+    register('--pantsd-log-dir', advanced=True, default=None,
+             help='The directory to log pantsd output to.')
+    register('--pantsd-fs-event-detection', advanced=True, type=bool,
+             removal_version='1.5.0.dev0',
+             removal_hint='This option is now implied by `--enable-pantsd`.',
+             help='Whether or not to use filesystem event detection.')
+    register('--pantsd-fs-event-workers', advanced=True, type=int, default=4,
+             help='The number of workers to use for the filesystem event service executor pool.')
+
+    # Watchman options.
+    register('--watchman-version', advanced=True, default='4.5.0', help='Watchman version.')
+    register('--watchman-supportdir', advanced=True, default='bin/watchman',
+             help='Find watchman binaries under this dir. Used as part of the path to lookup '
+                  'the binary with --binary-util-baseurls and --pants-bootstrapdir.')
+    register('--watchman-startup-timeout', type=float, advanced=True, default=30.0,
+             help='The watchman socket timeout (in seconds) for the initial `watch-project` command. '
+                  'This may need to be set higher for larger repos due to watchman startup cost.')
+    register('--watchman-socket-timeout', type=float, advanced=True, default=5.0,
+             help='The watchman client socket timeout in seconds.')
+    register('--watchman-socket-path', type=str, advanced=True, default=None,
+             help='The path to the watchman UNIX socket. This can be overridden if the default '
+                  'absolute path length exceeds the maximum allowed by the OS.')
+
   @classmethod
   def register_options(cls, register):
     """Register options not tied to any particular task or subsystem."""
@@ -187,23 +239,6 @@ def register_options(cls, register):
              help="Constrain what Python interpreters to use.  Uses Requirement format from "
                   "pkg_resources, e.g. 'CPython>=2.7,<3' or 'PyPy'. By default, no constraints "
                   "are used.  Multiple constraints may be added.  They will be ORed together.")
-    register('--exclude-target-regexp', advanced=True, type=list, default=[],
-             metavar='<regexp>',
-             help='Exclude target roots that match these regexes.')
-    # Relative pants_distdir to buildroot. Requires --pants-distdir to be bootstrapped above first.
-    # e.g. '/dist/'
-    rel_distdir = '/{}/'.format(os.path.relpath(register.bootstrap.pants_distdir, get_buildroot()))
-    register('--build-ignore', advanced=True, type=list, fromfile=True,
-             default=['.*/', rel_distdir, 'bower_components/', 'node_modules/', '*.egg-info/'],
-             help='Paths to ignore when identifying BUILD files. '
-                  'This does not affect any other filesystem operations. '
-                  'Patterns use the gitignore pattern syntax (https://git-scm.com/docs/gitignore).')
-    register('--pants-ignore', advanced=True, type=list, fromfile=True, default=['.*/', rel_distdir],
-             help='Paths to ignore for all filesystem operations performed by pants '
-                  '(e.g. BUILD file scanning, glob matching, etc). '
-                  'Patterns use the gitignore syntax (https://git-scm.com/docs/gitignore). '
-                  'This currently only affects the v2 engine. '
-                  'To experiment with v2 engine, try --enable-v2-engine option.')
     register('--fail-fast', advanced=True, type=bool, recursive=True,
              help='Exit as quickly as possible on error, rather than attempting to continue '
                   'to process the non-erroneous subset of the input.')
@@ -224,6 +259,3 @@ def register_options(cls, register):
     register('--lock', advanced=True, type=bool, default=True,
              help='Use a global lock to exclude other versions of pants from running during '
                   'critical operations.')
-    register('--subproject-roots', type=list, advanced=True, fromfile=True, default=[],
-             help='Paths that correspond with build roots for any subproject that this '
-                  'project depends on.')
diff --git a/src/python/pants/pantsd/BUILD b/src/python/pants/pantsd/BUILD
index 5499547678c..1c69d93ac51 100644
--- a/src/python/pants/pantsd/BUILD
+++ b/src/python/pants/pantsd/BUILD
@@ -7,7 +7,6 @@ python_library(
   dependencies = [
     '3rdparty/python:psutil',
     'src/python/pants/base:build_environment',
-    'src/python/pants/pantsd/subsystem:subprocess',
     'src/python/pants/util:dirutil',
     'src/python/pants/util:process_handler',
   ]
@@ -35,6 +34,16 @@ python_library(
   ]
 )
 
+python_library(
+  name = 'watchman_launcher',
+  sources = ['watchman_launcher.py'],
+  dependencies = [
+    ':watchman',
+    'src/python/pants/binaries:binary_util',
+    'src/python/pants/util:memo',
+  ]
+)
+
 python_library(
   name = 'watchman_client',
   sources = ['watchman_client.py'],
@@ -54,3 +63,21 @@ python_library(
     ':process_manager',
   ]
 )
+
+python_library(
+  name = 'pants_daemon_launcher',
+  sources = ['pants_daemon_launcher.py'],
+  dependencies = [
+    ':pants_daemon',
+    ':watchman_launcher',
+    'src/python/pants/pantsd/service:fs_event_service',
+    'src/python/pants/pantsd/service:pailgun_service',
+    'src/python/pants/pantsd/service:scheduler_service',
+    'src/python/pants/binaries:binary_util',
+    'src/python/pants/base:build_environment',
+    'src/python/pants/engine:native',
+    'src/python/pants/init',
+    'src/python/pants/process',
+    'src/python/pants/util:memo'
+  ]
+)
diff --git a/src/python/pants/pantsd/pailgun_server.py b/src/python/pants/pantsd/pailgun_server.py
index 6f81d10e30b..add43f2acaf 100644
--- a/src/python/pants/pantsd/pailgun_server.py
+++ b/src/python/pants/pantsd/pailgun_server.py
@@ -132,8 +132,7 @@ def process_request(self, request, client_address):
 
     try:
       # Attempt to handle a request with the handler under the context_lock.
-      with self._context_lock():
-        handler.handle_request()
+      handler.handle_request()
     except Exception as e:
       # If that fails, (synchronously) handle the error with the error handler sans-fork.
       try:
diff --git a/src/python/pants/pantsd/pants_daemon.py b/src/python/pants/pantsd/pants_daemon.py
index d3e5fb0a17a..9666cb285af 100644
--- a/src/python/pants/pantsd/pants_daemon.py
+++ b/src/python/pants/pantsd/pants_daemon.py
@@ -97,12 +97,13 @@ def set_socket_map(self, socket_map):
 
   def shutdown(self, service_thread_map):
     """Gracefully terminate all services and kill the main PantsDaemon loop."""
-    for service, service_thread in service_thread_map.items():
-      self._logger.info('terminating pantsd service: {}'.format(service))
-      service.terminate()
-      service_thread.join()
-    self._logger.info('terminating pantsd')
-    self._kill_switch.set()
+    with self._lock:
+      for service, service_thread in service_thread_map.items():
+        self._logger.info('terminating pantsd service: {}'.format(service))
+        service.terminate()
+        service_thread.join()
+      self._logger.info('terminating pantsd')
+      self._kill_switch.set()
 
   @staticmethod
   def _close_fds():
@@ -164,6 +165,9 @@ def _run_services(self, services):
         self.shutdown(service_thread_map)
         raise self.StartupFailure('service {} failed to start, shutting down!'.format(service))
 
+    # Once all services are started, write our pid.
+    self.write_pid()
+
     # Monitor services.
     while not self.is_killed:
       for service, service_thread in service_thread_map.items():
diff --git a/src/python/pants/pantsd/pants_daemon_launcher.py b/src/python/pants/pantsd/pants_daemon_launcher.py
new file mode 100644
index 00000000000..d3f4fa10c13
--- /dev/null
+++ b/src/python/pants/pantsd/pants_daemon_launcher.py
@@ -0,0 +1,143 @@
+# coding=utf-8
+# Copyright 2015 Pants project contributors (see CONTRIBUTORS.md).
+# Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+from __future__ import (absolute_import, division, generators, nested_scopes, print_function,
+                        unicode_literals, with_statement)
+
+import logging
+import os
+
+from pants.base.build_environment import get_buildroot
+from pants.bin.daemon_pants_runner import DaemonExiter, DaemonPantsRunner
+from pants.engine.native import Native
+from pants.init.target_roots import TargetRoots
+from pants.init.util import clean_global_runtime_state
+from pants.pantsd.pants_daemon import PantsDaemon
+from pants.pantsd.service.fs_event_service import FSEventService
+from pants.pantsd.service.pailgun_service import PailgunService
+from pants.pantsd.service.scheduler_service import SchedulerService
+from pants.pantsd.watchman_launcher import WatchmanLauncher
+from pants.process.lock import OwnerPrintingInterProcessFileLock
+from pants.util.memo import testable_memoized_property
+
+
+class PantsDaemonLauncher(object):
+  """An object that manages the configuration and lifecycle of pantsd."""
+
+  def __init__(self, bootstrap_options, engine_initializer=None):
+    """
+    :param Options bootstrap_options: An Options object containing the bootstrap options.
+    :param class engine_initializer: The class representing the EngineInitializer. Only required
+                                     for startup.
+    """
+    self._bootstrap_options = bootstrap_options
+    self._engine_initializer = engine_initializer
+
+    self._pailgun_host = bootstrap_options.pantsd_pailgun_host
+    self._pailgun_port = bootstrap_options.pantsd_pailgun_port
+    self._log_dir = bootstrap_options.pantsd_log_dir
+    self._fs_event_workers = bootstrap_options.pantsd_fs_event_workers
+    self._pants_workdir = bootstrap_options.pants_workdir
+    self._log_level = bootstrap_options.level.upper()
+    self._pants_ignore_patterns = bootstrap_options.pants_ignore
+    self._build_ignore_patterns = bootstrap_options.build_ignore
+    self._exclude_target_regexp = bootstrap_options.exclude_target_regexp
+    self._subproject_roots = bootstrap_options.subproject_roots
+    self._metadata_base_dir = bootstrap_options.pants_subprocessdir
+
+    # TODO: https://github.com/pantsbuild/pants/issues/3479
+    self._build_root = get_buildroot()
+    self._native = Native.create(bootstrap_options)
+    self._logger = logging.getLogger(__name__)
+    self._lock = OwnerPrintingInterProcessFileLock(
+      os.path.join(self._build_root, '.pantsd.startup')
+    )
+
+  @testable_memoized_property
+  def pantsd(self):
+    return PantsDaemon(
+      self._build_root,
+      self._pants_workdir,
+      self._log_level,
+      self._native,
+      self._log_dir,
+      reset_func=clean_global_runtime_state,
+      metadata_base_dir=self._metadata_base_dir
+    )
+
+  @testable_memoized_property
+  def watchman_launcher(self):
+    return WatchmanLauncher.create(self._bootstrap_options)
+
+  def _setup_services(self, watchman):
+    """Initialize pantsd services.
+
+    :returns: A tuple of (`tuple` service_instances, `dict` port_map).
+    """
+    legacy_graph_helper = self._engine_initializer.setup_legacy_graph(
+      self._pants_ignore_patterns,
+      self._pants_workdir,
+      native=self._native,
+      build_ignore_patterns=self._build_ignore_patterns,
+      exclude_target_regexps=self._exclude_target_regexp,
+      subproject_roots=self._subproject_roots,
+    )
+
+    fs_event_service = FSEventService(watchman, self._build_root, self._fs_event_workers)
+    scheduler_service = SchedulerService(fs_event_service, legacy_graph_helper)
+    pailgun_service = PailgunService(
+      bind_addr=(self._pailgun_host, self._pailgun_port),
+      exiter_class=DaemonExiter,
+      runner_class=DaemonPantsRunner,
+      target_roots_class=TargetRoots,
+      scheduler_service=scheduler_service
+    )
+
+    return (
+      # Use the schedulers reentrant lock as the daemon's global lock.
+      legacy_graph_helper.scheduler.lock,
+      # Services.
+      (fs_event_service, scheduler_service, pailgun_service),
+      # Port map.
+      dict(pailgun=pailgun_service.pailgun_port)
+    )
+
+  def _launch_pantsd(self):
+    # Launch Watchman (if so configured).
+    watchman = self.watchman_launcher.maybe_launch()
+
+    # Initialize pantsd services.
+    lock, services, port_map = self._setup_services(watchman)
+
+    # Setup and fork pantsd.
+    self.pantsd.set_lock(lock)
+    self.pantsd.set_services(services)
+    self.pantsd.set_socket_map(port_map)
+    # Defer pid writing until the daemon has fully spawned.
+    self.pantsd.daemonize(write_pid=False)
+
+    # Wait up to 10 seconds for pantsd to write its pidfile so we can display the pid to the user.
+    self.pantsd.await_pid(10)
+
+  def maybe_launch(self):
+    """Launches pantsd if not already running.
+
+    :returns: The port that pantsd is listening on.
+    :rtype: int
+    """
+    self._logger.debug('acquiring lock: {}'.format(self._lock))
+    with self._lock:
+      if not self.pantsd.is_alive():
+        self._logger.debug('launching pantsd')
+        self._launch_pantsd()
+      listening_port = self.pantsd.read_named_socket('pailgun', int)
+      pantsd_pid = self.pantsd.pid
+    self._logger.debug('released lock: {}'.format(self._lock))
+    self._logger.debug('pantsd is running at pid {}, pailgun port is {}'
+                       .format(pantsd_pid, listening_port))
+    return listening_port
+
+  def terminate(self):
+    self.pantsd.terminate()
+    self.watchman_launcher.terminate()
diff --git a/src/python/pants/pantsd/process_manager.py b/src/python/pants/pantsd/process_manager.py
index e49593db944..a545691b756 100644
--- a/src/python/pants/pantsd/process_manager.py
+++ b/src/python/pants/pantsd/process_manager.py
@@ -15,7 +15,7 @@
 import psutil
 
 from pants.base.build_environment import get_buildroot
-from pants.pantsd.subsystem.subprocess import Subprocess
+from pants.init.subprocess import Subprocess
 from pants.util.dirutil import read_file, rm_rf, safe_file_dump, safe_mkdir
 from pants.util.process_handler import subprocess
 
@@ -301,8 +301,9 @@ def await_socket(self, timeout):
     """Wait up to a given timeout for a process to write socket info."""
     return self.await_metadata_by_name(self._name, 'socket', timeout, self._socket_type)
 
-  def write_pid(self, pid):
+  def write_pid(self, pid=None):
     """Write the current processes PID to the pidfile location"""
+    pid = pid or os.getpid()
     self.write_metadata_by_name(self._name, 'pid', str(pid))
 
   def write_socket(self, socket_info):
@@ -313,6 +314,10 @@ def write_named_socket(self, socket_name, socket_info):
     """A multi-tenant, named alternative to ProcessManager.write_socket()."""
     self.write_metadata_by_name(self._name, 'socket_{}'.format(socket_name), str(socket_info))
 
+  def read_named_socket(self, socket_name, socket_type):
+    """A multi-tenant, named alternative to ProcessManager.socket."""
+    return self.read_metadata_by_name(self._name, 'socket_{}'.format(socket_name), socket_type)
+
   def _as_process(self):
     """Returns a psutil `Process` object wrapping our pid.
 
diff --git a/src/python/pants/pantsd/service/scheduler_service.py b/src/python/pants/pantsd/service/scheduler_service.py
index a5af2162296..cacf4bc1c7a 100644
--- a/src/python/pants/pantsd/service/scheduler_service.py
+++ b/src/python/pants/pantsd/service/scheduler_service.py
@@ -80,13 +80,11 @@ def _process_event_queue(self):
       self._logger.debug('processing {} files for subscription {} (first_event={})'
                          .format(len(files), subscription, is_initial_event))
 
-      if is_initial_event:
-        # Once we've seen the initial watchman event, set the internal `Event`.
-        self._ready.set()
-      else:
-        # The first watchman event is a listing of all files - ignore it.
+      # The first watchman event is a listing of all files - ignore it.
+      if not is_initial_event:
         self._handle_batch_event(files)
 
+    if not self._ready.is_set(): self._ready.set()
     self._event_queue.task_done()
 
   def warm_product_graph(self, spec_roots):
diff --git a/src/python/pants/pantsd/subsystem/BUILD b/src/python/pants/pantsd/subsystem/BUILD
deleted file mode 100644
index 3a69b9c073d..00000000000
--- a/src/python/pants/pantsd/subsystem/BUILD
+++ /dev/null
@@ -1,23 +0,0 @@
-# Copyright 2015 Pants project contributors (see CONTRIBUTORS.md).
-# Licensed under the Apache License, Version 2.0 (see LICENSE).
-
-python_library(
-  name = 'subprocess',
-  sources = ['subprocess.py'],
-  dependencies = [
-    'src/python/pants/option',
-    'src/python/pants/subsystem',
-  ]
-)
-
-python_library(
-  name = 'watchman_launcher',
-  sources = ['watchman_launcher.py'],
-  dependencies = [
-    ':subprocess',
-    'src/python/pants/binaries:binary_util',
-    'src/python/pants/pantsd:watchman',
-    'src/python/pants/subsystem:subsystem',
-    'src/python/pants/util:memo',
-  ]
-)
diff --git a/src/python/pants/pantsd/subsystem/__init__.py b/src/python/pants/pantsd/subsystem/__init__.py
deleted file mode 100644
index e69de29bb2d..00000000000
diff --git a/src/python/pants/pantsd/watchman.py b/src/python/pants/pantsd/watchman.py
index 368639c17ff..bed189ba9a4 100644
--- a/src/python/pants/pantsd/watchman.py
+++ b/src/python/pants/pantsd/watchman.py
@@ -20,6 +20,9 @@
 class Watchman(ProcessManager):
   """Watchman process manager and helper class."""
 
+  class WatchmanCrash(Exception):
+    """Raised when Watchman crashes."""
+
   STARTUP_TIMEOUT_SECONDS = 30.0
   SOCKET_TIMEOUT_SECONDS = 5.0
 
@@ -77,7 +80,6 @@ def _normalize_watchman_path(self, watchman_path):
     return os.path.abspath(watchman_path)
 
   def _maybe_init_metadata(self):
-    self._logger.debug('ensuring creation of directory: {}'.format(self._watchman_work_dir))
     safe_mkdir(self._watchman_work_dir)
     # Initialize watchman with an empty, but valid statefile so it doesn't complain on startup.
     safe_file_dump(self._state_file, '{}')
@@ -138,6 +140,15 @@ def launch(self):
     self.write_pid(pid)
     self.write_socket(self._sock_file)
 
+  def _attempt_set_timeout(self, timeout):
+    """Sets a timeout on the inner watchman client's socket."""
+    try:
+      self.client.setTimeout(timeout)
+    except Exception:
+      self._logger.debug('failed to set post-startup watchman timeout to %s', self._timeout)
+    else:
+      self._logger.debug('set post-startup watchman timeout to %s', self._timeout)
+
   def watch_project(self, path):
     """Issues the watch-project command to watchman to begin watching the buildroot.
 
@@ -147,8 +158,7 @@ def watch_project(self, path):
     try:
       return self.client.query('watch-project', os.path.realpath(path))
     finally:
-      self.client.setTimeout(self._timeout)
-      self._logger.debug('setting post-startup watchman timeout to %s', self._timeout)
+      self._attempt_set_timeout(self._timeout)
 
   def subscribed(self, build_root, handlers):
     """Bulk subscribe generator for StreamableWatchmanClient.
@@ -164,13 +174,16 @@ def subscribed(self, build_root, handlers):
 
     self._logger.debug('watchman command_list is: {}'.format(command_list))
 
-    for event in self.client.stream_query(command_list):
-      if event is None:
-        yield None, None
-      elif 'subscribe' in event:
-        self._logger.info('confirmed watchman subscription: {}'.format(event))
-        yield None, None
-      elif 'subscription' in event:
-        yield event.get('subscription'), event
-      else:
-        self._logger.warning('encountered non-subscription event: {}'.format(event))
+    try:
+      for event in self.client.stream_query(command_list):
+        if event is None:
+          yield None, None
+        elif 'subscribe' in event:
+          self._logger.info('confirmed watchman subscription: {}'.format(event))
+          yield None, None
+        elif 'subscription' in event:
+          yield event.get('subscription'), event
+        else:
+          self._logger.warning('encountered non-subscription event: {}'.format(event))
+    except self.client.WatchmanError as e:
+      raise self.WatchmanCrash(e)
diff --git a/src/python/pants/pantsd/watchman_client.py b/src/python/pants/pantsd/watchman_client.py
index 81c0e6671b4..1ccd546083d 100644
--- a/src/python/pants/pantsd/watchman_client.py
+++ b/src/python/pants/pantsd/watchman_client.py
@@ -14,6 +14,7 @@
 class StreamableWatchmanClient(pywatchman.client):
   """A watchman client subclass that provides for interruptable unilateral queries."""
 
+  WatchmanError = pywatchman.WatchmanError
   SocketTimeout = pywatchman.SocketTimeout
 
   def stream_query(self, commands):
diff --git a/src/python/pants/pantsd/subsystem/watchman_launcher.py b/src/python/pants/pantsd/watchman_launcher.py
similarity index 56%
rename from src/python/pants/pantsd/subsystem/watchman_launcher.py
rename to src/python/pants/pantsd/watchman_launcher.py
index bc15cae8a49..3417a6df704 100644
--- a/src/python/pants/pantsd/subsystem/watchman_launcher.py
+++ b/src/python/pants/pantsd/watchman_launcher.py
@@ -8,52 +8,39 @@
 import logging
 
 from pants.binaries.binary_util import BinaryUtil
-from pants.pantsd.subsystem.subprocess import Subprocess
 from pants.pantsd.watchman import Watchman
-from pants.subsystem.subsystem import Subsystem
 from pants.util.memo import testable_memoized_property
 
 
 class WatchmanLauncher(object):
-  """A subsystem that encapsulates access to Watchman."""
+  """An object that manages the configuration and lifecycle of Watchman."""
 
-  class Factory(Subsystem):
-    options_scope = 'watchman'
-
-    @classmethod
-    def subsystem_dependencies(cls):
-      return (BinaryUtil.Factory, Subprocess.Factory)
-
-    @classmethod
-    def register_options(cls, register):
-      register('--version', advanced=True, default='4.9.0',
-               help='Watchman version.')
-      register('--supportdir', advanced=True, default='bin/watchman',
-               help='Find watchman binaries under this dir. Used as part of the path to lookup '
-                    'the binary with --binary-util-baseurls and --pants-bootstrapdir.')
-      register('--startup-timeout', type=float, advanced=True, default=Watchman.STARTUP_TIMEOUT_SECONDS,
-               help='The watchman socket timeout (in seconds) for the initial `watch-project` command. '
-                    'This may need to be set higher for larger repos due to watchman startup cost.')
-      register('--socket-timeout', type=float, advanced=True, default=Watchman.SOCKET_TIMEOUT_SECONDS,
-               help='The watchman client socket timeout (in seconds).')
-      register('--socket-path', type=str, advanced=True, default=None,
-               help='The path to the watchman UNIX socket. This can be overridden if the default '
-                    'absolute path length exceeds the maximum allowed by the OS.')
+  @classmethod
+  def create(cls, bootstrap_options):
+    """
+    :param Options bootstrap_options: The bootstrap options bag.
+    """
+    binary_util = BinaryUtil(
+      bootstrap_options.binaries_baseurls,
+      bootstrap_options.binaries_fetch_timeout_secs,
+      bootstrap_options.pants_bootstrapdir,
+      bootstrap_options.binaries_path_by_id
+    )
 
-    def create(self):
-      binary_util = BinaryUtil.Factory.create()
-      options = self.get_options()
-      return WatchmanLauncher(binary_util,
-                              options.pants_workdir,
-                              options.level,
-                              options.version,
-                              options.supportdir,
-                              options.startup_timeout,
-                              options.socket_timeout,
-                              options.socket_path)
+    return WatchmanLauncher(
+      binary_util,
+      bootstrap_options.pants_workdir,
+      bootstrap_options.level,
+      bootstrap_options.watchman_version,
+      bootstrap_options.watchman_supportdir,
+      bootstrap_options.watchman_startup_timeout,
+      bootstrap_options.watchman_socket_timeout,
+      bootstrap_options.watchman_socket_path,
+      bootstrap_options.pants_subprocessdir
+    )
 
   def __init__(self, binary_util, workdir, log_level, watchman_version, watchman_supportdir,
-               startup_timeout, socket_timeout, socket_path_override=None):
+               startup_timeout, socket_timeout, socket_path_override=None, metadata_base_dir=None):
     """
     :param binary_util: The BinaryUtil subsystem instance for binary retrieval.
     :param workdir: The current pants workdir.
@@ -62,6 +49,7 @@ def __init__(self, binary_util, workdir, log_level, watchman_version, watchman_s
     :param watchman_supportdir: The supportdir for BinaryUtil.
     :param socket_timeout: The watchman client socket timeout (in seconds).
     :param socket_path_override: The overridden target path of the watchman socket, if any.
+    :param metadata_base_dir: The ProcessManager metadata base directory.
     """
     self._binary_util = binary_util
     self._workdir = workdir
@@ -72,7 +60,7 @@ def __init__(self, binary_util, workdir, log_level, watchman_version, watchman_s
     self._socket_path_override = socket_path_override
     self._log_level = log_level
     self._logger = logging.getLogger(__name__)
-    self._watchman = None
+    self._metadata_base_dir = metadata_base_dir
 
   @staticmethod
   def _convert_log_level(level):
@@ -88,12 +76,15 @@ def watchman(self):
     watchman_binary = self._binary_util.select_binary(self._watchman_supportdir,
                                                       self._watchman_version,
                                                       'watchman')
-    return Watchman(watchman_binary,
-                    self._workdir,
-                    self._convert_log_level(self._log_level),
-                    self._startup_timeout,
-                    self._socket_timeout,
-                    self._socket_path_override)
+    return Watchman(
+      watchman_binary,
+      self._workdir,
+      self._convert_log_level(self._log_level),
+      self._startup_timeout,
+      self._socket_timeout,
+      self._socket_path_override,
+      metadata_base_dir=self._metadata_base_dir
+    )
 
   def maybe_launch(self):
     if not self.watchman.is_alive():
@@ -106,6 +97,7 @@ def maybe_launch(self):
 
     self._logger.debug('watchman is running, pid={pid} socket={socket}'
                        .format(pid=self.watchman.pid, socket=self.watchman.socket))
+
     return self.watchman
 
   def terminate(self):
diff --git a/src/python/pants/util/BUILD b/src/python/pants/util/BUILD
index 29f30a16302..697b297c0c9 100644
--- a/src/python/pants/util/BUILD
+++ b/src/python/pants/util/BUILD
@@ -17,6 +17,11 @@ python_library(
    ],
 )
 
+python_library(
+  name = 'collections',
+  sources = ['collections.py'],
+)
+
 python_library(
   name = 'desktop',
   sources = ['desktop.py'],
diff --git a/src/python/pants/util/collections.py b/src/python/pants/util/collections.py
new file mode 100644
index 00000000000..c2e45161cd9
--- /dev/null
+++ b/src/python/pants/util/collections.py
@@ -0,0 +1,11 @@
+# coding=utf-8
+# Copyright 2017 Pants project contributors (see CONTRIBUTORS.md).
+# Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+from __future__ import (absolute_import, division, generators, nested_scopes, print_function,
+                        unicode_literals, with_statement)
+
+
+def combined_dict(*dicts):
+  """Combine one or more dicts into a new, unified dict (dicts to the right take precedence)."""
+  return {k: v for d in dicts for k, v in d.items()}
diff --git a/tests/python/pants_test/base_test.py b/tests/python/pants_test/base_test.py
index 8361400d1ab..b398ce27ff2 100644
--- a/tests/python/pants_test/base_test.py
+++ b/tests/python/pants_test/base_test.py
@@ -25,6 +25,7 @@
 from pants.build_graph.mutable_build_graph import MutableBuildGraph
 from pants.build_graph.target import Target
 from pants.init.util import clean_global_runtime_state
+from pants.option.options_bootstrapper import OptionsBootstrapper
 from pants.source.source_root import SourceRootConfig
 from pants.subsystem.subsystem import Subsystem
 from pants.util.dirutil import safe_mkdir, safe_open, safe_rmtree
@@ -435,3 +436,12 @@ def assertPrefixEqual(self, expected, actual_iter):
     :API: public
     """
     self.assertEqual(expected, list(itertools.islice(actual_iter, len(expected))))
+
+  def get_bootstrap_options(self, cli_options=()):
+    """Retrieves bootstrap options.
+
+    :param cli_options: An iterable of CLI flags to pass as arguments to `OptionsBootstrapper`.
+    """
+    # Can't parse any options without a pants.ini.
+    self.create_file('pants.ini')
+    return OptionsBootstrapper(args=cli_options).get_bootstrap_options().for_global_scope()
diff --git a/tests/python/pants_test/pantsd/subsystem/test_subprocess.py b/tests/python/pants_test/init/test_subprocess.py
similarity index 92%
rename from tests/python/pants_test/pantsd/subsystem/test_subprocess.py
rename to tests/python/pants_test/init/test_subprocess.py
index 745e04ceabb..19a0f043c0b 100644
--- a/tests/python/pants_test/pantsd/subsystem/test_subprocess.py
+++ b/tests/python/pants_test/init/test_subprocess.py
@@ -5,7 +5,7 @@
 from __future__ import (absolute_import, division, generators, nested_scopes, print_function,
                         unicode_literals, with_statement)
 
-from pants.pantsd.subsystem.subprocess import Subprocess
+from pants.init.subprocess import Subprocess
 from pants_test.base_test import BaseTest
 from pants_test.subsystem.subsystem_util import global_subsystem_instance
 
diff --git a/tests/python/pants_test/option/test_options_integration.py b/tests/python/pants_test/option/test_options_integration.py
index 15da2bc1653..ff59ae7da81 100644
--- a/tests/python/pants_test/option/test_options_integration.py
+++ b/tests/python/pants_test/option/test_options_integration.py
@@ -291,21 +291,3 @@ def test_pants_ignore_option(self):
       self.assertIn("pants_ignore = ['.*/', '/dist/', 'some/random/dir'] (from CONFIG in {})"
                     .format(config_path),
                     pants_run.stdout_data)
-
-  @ensure_engine
-  def test_pants_ignore_option_non_default_dist_dir(self):
-    with temporary_dir(root_dir=os.path.abspath('.')) as tempdir:
-      config_path = os.path.relpath(os.path.join(tempdir, 'config.ini'))
-      with open(config_path, 'w+') as f:
-        f.write(dedent("""
-          [GLOBAL]
-          pants_ignore: +['some/random/dir']
-          pants_distdir: some/other/dist/dir
-        """))
-      pants_run = self.run_pants(['--pants-config-files={}'.format(config_path),
-                                  '--no-colors',
-                                  'options'])
-      self.assert_success(pants_run)
-      self.assertIn("pants_ignore = ['.*/', '/some/other/dist/dir/', 'some/random/dir'] "
-                    "(from CONFIG in {})".format(config_path),
-                    pants_run.stdout_data)
diff --git a/tests/python/pants_test/pantsd/BUILD b/tests/python/pants_test/pantsd/BUILD
index 3a8cfa65bd2..150e86f38cc 100644
--- a/tests/python/pants_test/pantsd/BUILD
+++ b/tests/python/pants_test/pantsd/BUILD
@@ -73,3 +73,25 @@ python_tests(
   tags = {'integration'},
   timeout = 120
 )
+
+python_tests(
+  name = 'pants_daemon_launcher',
+  sources = ['test_pants_daemon_launcher.py'],
+  coverage = ['pants.pantsd.pants_daemon_launcher'],
+  dependencies = [
+    'tests/python/pants_test/pantsd:test_deps',
+    'tests/python/pants_test/subsystem:subsystem_utils',
+    'src/python/pants/pantsd:pants_daemon_launcher'
+  ]
+)
+
+python_tests(
+  name = 'watchman_launcher',
+  sources = ['test_watchman_launcher.py'],
+  coverage = ['pants.pantsd.watchman_launcher'],
+  dependencies = [
+    'tests/python/pants_test/pantsd:test_deps',
+    'tests/python/pants_test/subsystem:subsystem_utils',
+    'src/python/pants/pantsd:watchman_launcher'
+  ]
+)
diff --git a/tests/python/pants_test/pantsd/subsystem/BUILD b/tests/python/pants_test/pantsd/subsystem/BUILD
deleted file mode 100644
index 1adc01a42e0..00000000000
--- a/tests/python/pants_test/pantsd/subsystem/BUILD
+++ /dev/null
@@ -1,24 +0,0 @@
-# Copyright 2015 Pants project contributors (see CONTRIBUTORS.md).
-# Licensed under the Apache License, Version 2.0 (see LICENSE).
-
-python_tests(
-  name = 'subprocess',
-  sources = ['test_subprocess.py'],
-  coverage = ['pants.pantsd.subsystem.subprocess'],
-  dependencies = [
-    'src/python/pants/pantsd/subsystem:subprocess',
-    'tests/python/pants_test/subsystem:subsystem_utils',
-    'tests/python/pants_test:base_test'
-  ]
-)
-
-python_tests(
-  name = 'watchman_launcher',
-  sources = ['test_watchman_launcher.py'],
-  coverage = ['pants.pantsd.subsystem.watchman_launcher'],
-  dependencies = [
-    'tests/python/pants_test/pantsd:test_deps',
-    'tests/python/pants_test/subsystem:subsystem_utils',
-    'src/python/pants/pantsd/subsystem:watchman_launcher'
-  ]
-)
diff --git a/tests/python/pants_test/pantsd/subsystem/__init__.py b/tests/python/pants_test/pantsd/subsystem/__init__.py
deleted file mode 100644
index e69de29bb2d..00000000000
diff --git a/tests/python/pants_test/pantsd/test_pants_daemon.py b/tests/python/pants_test/pantsd/test_pants_daemon.py
index 254c88a10ab..a0843e4b80d 100644
--- a/tests/python/pants_test/pantsd/test_pants_daemon.py
+++ b/tests/python/pants_test/pantsd/test_pants_daemon.py
@@ -6,6 +6,7 @@
                         unicode_literals, with_statement)
 
 import logging
+import threading
 
 import mock
 
@@ -44,6 +45,7 @@ def test_flush(self):
 class PantsDaemonTest(BaseTest):
   def setUp(self):
     super(PantsDaemonTest, self).setUp()
+    lock = threading.RLock()
     self.pantsd = PantsDaemon('test_buildroot',
                               'test_work_dir',
                               logging.INFO,
@@ -52,6 +54,7 @@ def setUp(self):
                               metadata_base_dir=self.subprocess_dir)
     self.pantsd.set_services([])
     self.pantsd.set_socket_map({})
+    self.pantsd.set_lock(lock)
 
     self.mock_killswitch = mock.Mock()
     self.pantsd._kill_switch = self.mock_killswitch
diff --git a/tests/python/pants_test/init/test_pants_daemon_launcher.py b/tests/python/pants_test/pantsd/test_pants_daemon_launcher.py
similarity index 79%
rename from tests/python/pants_test/init/test_pants_daemon_launcher.py
rename to tests/python/pants_test/pantsd/test_pants_daemon_launcher.py
index 06bdb57602e..203148c0aaf 100644
--- a/tests/python/pants_test/init/test_pants_daemon_launcher.py
+++ b/tests/python/pants_test/pantsd/test_pants_daemon_launcher.py
@@ -7,19 +7,18 @@
 
 import mock
 
-from pants.init.pants_daemon_launcher import PantsDaemonLauncher
 from pants.pantsd.pants_daemon import PantsDaemon
-from pants.pantsd.subsystem.watchman_launcher import WatchmanLauncher
+from pants.pantsd.pants_daemon_launcher import PantsDaemonLauncher
+from pants.pantsd.watchman_launcher import WatchmanLauncher
 from pants_test.base_test import BaseTest
-from pants_test.subsystem.subsystem_util import global_subsystem_instance
 
 
 class PantsDaemonLauncherTest(BaseTest):
   PDL_PATCH_OPTS = dict(autospec=True, spec_set=True, return_value=(None, None, None))
 
-  def pants_daemon_launcher(self):
-    factory = global_subsystem_instance(PantsDaemonLauncher.Factory)
-    pdl = factory.create(None)
+  def pants_daemon_launcher(self, cli_options=()):
+    bootstrap_options = self.get_bootstrap_options(cli_options)
+    pdl = PantsDaemonLauncher(bootstrap_options)
     pdl.pantsd = self.mock_pantsd
     pdl.watchman_launcher = self.mock_watchman_launcher
     return pdl
@@ -43,9 +42,8 @@ def test_maybe_launch(self, mock_setup_services):
   @mock.patch.object(PantsDaemonLauncher, '_setup_services', **PDL_PATCH_OPTS)
   def test_maybe_launch_already_alive(self, mock_setup_services):
     self.mock_pantsd.is_alive.return_value = True
-    #options = {'default': {'pantsd_enabled': 'true'}}
 
-    pdl = self.pants_daemon_launcher()
+    pdl = self.pants_daemon_launcher(['--pantsd-enabled'])
     pdl.maybe_launch()
 
     self.assertEqual(mock_setup_services.call_count, 0)
diff --git a/tests/python/pants_test/pantsd/test_pantsd_integration.py b/tests/python/pants_test/pantsd/test_pantsd_integration.py
index b620a0a643c..17d4cc8912a 100644
--- a/tests/python/pants_test/pantsd/test_pantsd_integration.py
+++ b/tests/python/pants_test/pantsd/test_pantsd_integration.py
@@ -23,12 +23,10 @@ def await_pantsd(self, timeout=10):
     self._pid = self.await_pid(timeout)
     self.assert_running()
 
-  def assert_running(self, sleep=0):
-    time.sleep(sleep)
+  def assert_running(self):
     assert self._pid is not None and self.is_alive(), 'pantsd should be running!'
 
-  def assert_stopped(self, sleep=0):
-    time.sleep(sleep)
+  def assert_stopped(self):
     assert self._pid is not None and self.is_dead(), 'pantsd should be stopped!'
 
 
@@ -41,19 +39,25 @@ def read_pantsd_log(workdir):
 
 class TestPantsDaemonIntegration(PantsRunIntegrationTest):
   @contextmanager
-  def pantsd_test_context(self, config=None):
+  def pantsd_test_context(self, log_level='info'):
     with self.temporary_workdir() as workdir_base:
       pid_dir = os.path.join(workdir_base, '.pids')
       workdir = os.path.join(workdir_base, '.workdir.pants.d')
-      pantsd_config = config or {}
-      pantsd_config.setdefault('GLOBAL', {'enable_pantsd': True,
-                                          'level': 'debug',
-                                          'pants_subprocessdir': pid_dir})
+      pantsd_config = {
+        'GLOBAL': {
+          'enable_pantsd': True,
+          # The absolute paths in CI can exceed the UNIX socket path limitation
+          # (>104-108 characters), so we override that here with a shorter path.
+          'watchman_socket_path': '/tmp/watchman.{}.sock'.format(os.getpid()),
+          'level': log_level,
+          'pants_subprocessdir': pid_dir
+        }
+      }
       checker = PantsDaemonMonitor(pid_dir)
       yield workdir, pantsd_config, checker
 
-  def test_pantsd_run(self):
-    with self.pantsd_test_context() as (workdir, pantsd_config, checker):
+  def test_pantsd_compile(self):
+    with self.pantsd_test_context('debug') as (workdir, pantsd_config, checker):
       # Explicitly kill any running pantsd instances for the current buildroot.
       self.assert_success(self.run_pants_with_workdir(['kill-pantsd'], workdir, pantsd_config))
       try:
@@ -61,10 +65,6 @@ def test_pantsd_run(self):
         self.assert_success(self.run_pants_with_workdir(['help'], workdir, pantsd_config))
         checker.await_pantsd()
 
-        # This run should execute via pantsd testing the end to end client/server.
-        self.assert_success(self.run_pants_with_workdir(['help-advanced'], workdir, pantsd_config))
-        checker.assert_running()
-
         # This tests a deeper pantsd-based run by actually invoking a full compile.
         self.assert_success(
           self.run_pants_with_workdir(
@@ -74,20 +74,16 @@ def test_pantsd_run(self):
         )
         checker.assert_running()
       finally:
-        # Explicitly kill pantsd (from a pantsd-launched runner).
-        self.assert_success(self.run_pants_with_workdir(['kill-pantsd'], workdir, pantsd_config))
-
-      checker.assert_stopped()
-
-      for line in read_pantsd_log(workdir):
-        print(line)
+        try:
+          for line in read_pantsd_log(workdir):
+            print(line)
+        finally:
+          # Explicitly kill pantsd (from a pantsd-launched runner).
+          self.assert_success(self.run_pants_with_workdir(['kill-pantsd'], workdir, pantsd_config))
+          checker.assert_stopped()
 
-  def test_pantsd_run_with_watchman(self):
-    # The absolute paths in CI can exceed the UNIX socket path limitation
-    # (>104-108 characters), so we override that here with a shorter path.
-    config = {'watchman': {'socket_path': '/tmp/watchman.{}.sock'.format(os.getpid())}}
-
-    with self.pantsd_test_context(config) as (workdir, pantsd_config, checker):
+  def test_pantsd_run(self):
+    with self.pantsd_test_context('debug') as (workdir, pantsd_config, checker):
       print('log: {}/pantsd/pantsd.log'.format(workdir))
       # Explicitly kill any running pantsd instances for the current buildroot.
       print('\nkill-pantsd')
@@ -98,43 +94,39 @@ def test_pantsd_run_with_watchman(self):
         self.assert_success(self.run_pants_with_workdir(['help'], workdir, pantsd_config))
         checker.await_pantsd()
 
-        # This run should execute via pantsd testing the end to end client/server.
-        print('list')
+        print('list 3rdparty:')
         self.assert_success(self.run_pants_with_workdir(['list', '3rdparty:'],
                                                         workdir,
                                                         pantsd_config))
-        checker.assert_running(3)
+        checker.assert_running()
 
-        print('list')
+        print('list :')
         self.assert_success(self.run_pants_with_workdir(['list', ':'],
                                                         workdir,
                                                         pantsd_config))
-        checker.assert_running(3)
+        checker.assert_running()
 
-        print('list')
-        self.assert_success(self.run_pants_with_workdir(['list', ':'],
+        print('list ::')
+        self.assert_success(self.run_pants_with_workdir(['list', '::'],
                                                         workdir,
                                                         pantsd_config))
-        checker.assert_running(3)
+        checker.assert_running()
 
         # And again using the cached BuildGraph.
-        print('cached list')
+        print('list ::')
         self.assert_success(self.run_pants_with_workdir(['list', '::'],
                                                         workdir,
                                                         pantsd_config))
-        checker.assert_running(3)
+        checker.assert_running()
       finally:
-        for line in read_pantsd_log(workdir):
-          print(line)
-
-        # Explicitly kill pantsd (from a pantsd-launched runner).
-        print('kill-pantsd')
-        self.assert_success(self.run_pants_with_workdir(['kill-pantsd'], workdir, pantsd_config))
-
-      checker.assert_stopped()
-
-      for line in read_pantsd_log(workdir):
-        print(line)
+        try:
+          for line in read_pantsd_log(workdir):
+            print(line)
+        finally:
+          # Explicitly kill pantsd (from a pantsd-launched runner).
+          print('kill-pantsd')
+          self.assert_success(self.run_pants_with_workdir(['kill-pantsd'], workdir, pantsd_config))
+          checker.assert_stopped()
 
       # Assert there were no warnings or errors thrown in the pantsd log.
       for line in read_pantsd_log(workdir):
@@ -159,22 +151,19 @@ def test_pantsd_broken_pipe(self):
       finally:
         # Explicitly kill pantsd (from a pantsd-launched runner).
         self.assert_success(self.run_pants_with_workdir(['kill-pantsd'], workdir, pantsd_config))
-
-      checker.assert_stopped()
+        checker.assert_stopped()
 
   def test_pantsd_stacktrace_dump(self):
-    config = {'watchman': {'socket_path': '/tmp/watchman.{}.sock'.format(os.getpid())}}
-    with self.pantsd_test_context(config) as (workdir, pantsd_config, checker):
+    with self.pantsd_test_context() as (workdir, pantsd_config, checker):
       print('log: {}/pantsd/pantsd.log'.format(workdir))
       # Explicitly kill any running pantsd instances for the current buildroot.
       self.assert_success(self.run_pants_with_workdir(['kill-pantsd'], workdir, pantsd_config))
       try:
         # Start pantsd implicitly via a throwaway invocation.
         self.assert_success(self.run_pants_with_workdir(['help'], workdir, pantsd_config))
-        checker.await_pantsd(3)
+        checker.await_pantsd()
 
         os.kill(checker.pid, signal.SIGUSR2)
-        checker.assert_running()
 
         # Wait for log flush.
         time.sleep(2)
@@ -183,10 +172,9 @@ def test_pantsd_stacktrace_dump(self):
       finally:
         # Explicitly kill pantsd (from a pantsd-launched runner).
         self.assert_success(self.run_pants_with_workdir(['kill-pantsd'], workdir, pantsd_config))
+        checker.assert_stopped()
 
-      checker.assert_stopped()
-
-  def test_pantsd_runner_dies_after_failed_run(self):
+  def test_pantsd_runner_doesnt_die_after_failed_run(self):
     with self.pantsd_test_context() as (workdir, pantsd_config, checker):
       self.assert_success(self.run_pants_with_workdir(['kill-pantsd'], workdir, pantsd_config))
       try:
@@ -201,15 +189,15 @@ def test_pantsd_runner_dies_after_failed_run(self):
             workdir,
             pantsd_config)
         )
+        checker.assert_running()
 
         # Check for no stray pantsd-runner prcesses.
         self.assertFalse(check_process_exists_by_command('pantsd-runner'))
-        
+
         # Assert pantsd is in a good functional state.
         self.assert_success(self.run_pants_with_workdir(['help'], workdir, pantsd_config))
-
+        checker.assert_running()
       finally:
         # Explicitly kill pantsd (from a pantsd-launched runner).
         self.assert_success(self.run_pants_with_workdir(['kill-pantsd'], workdir, pantsd_config))
-
-      checker.assert_stopped()
+        checker.assert_stopped()
diff --git a/tests/python/pants_test/pantsd/subsystem/test_watchman_launcher.py b/tests/python/pants_test/pantsd/test_watchman_launcher.py
similarity index 83%
rename from tests/python/pants_test/pantsd/subsystem/test_watchman_launcher.py
rename to tests/python/pants_test/pantsd/test_watchman_launcher.py
index 7e162367acd..8b80088ca91 100644
--- a/tests/python/pants_test/pantsd/subsystem/test_watchman_launcher.py
+++ b/tests/python/pants_test/pantsd/test_watchman_launcher.py
@@ -7,16 +7,15 @@
 
 import mock
 
-from pants.pantsd.subsystem.watchman_launcher import WatchmanLauncher
 from pants.pantsd.watchman import Watchman
+from pants.pantsd.watchman_launcher import WatchmanLauncher
 from pants_test.base_test import BaseTest
-from pants_test.subsystem.subsystem_util import global_subsystem_instance
 
 
 class TestWatchmanLauncher(BaseTest):
-  def watchman_launcher(self, options=None):
-    options = options or {}
-    return global_subsystem_instance(WatchmanLauncher.Factory, options=options).create()
+  def watchman_launcher(self, cli_options=()):
+    bootstrap_options = self.get_bootstrap_options(cli_options)
+    return WatchmanLauncher.create(bootstrap_options)
 
   def create_mock_watchman(self, is_alive):
     mock_watchman = mock.create_autospec(Watchman, spec_set=False)
@@ -62,6 +61,6 @@ def test_watchman_property(self):
 
   def test_watchman_socket_path(self):
     expected_path = '/a/shorter/path'
-    options = {WatchmanLauncher.Factory.options_scope: {'socket_path': expected_path}}
+    options = ['--watchman-socket-path={}'.format(expected_path)]
     wl = self.watchman_launcher(options)
     self.assertEquals(wl.watchman._sock_file, expected_path)
diff --git a/tests/python/pants_test/util/BUILD b/tests/python/pants_test/util/BUILD
index a8672058cc8..c113e60128b 100644
--- a/tests/python/pants_test/util/BUILD
+++ b/tests/python/pants_test/util/BUILD
@@ -10,6 +10,15 @@ python_tests(
   ],
 )
 
+python_tests(
+  name = 'collections',
+  sources = ['test_collections.py'],
+  coverage = ['pants.util.collections'],
+  dependencies = [
+    'src/python/pants/util:collections',
+  ]
+)
+
 python_tests(
   name = 'contextutil',
   sources = ['test_contextutil.py'],
diff --git a/tests/python/pants_test/util/test_collections.py b/tests/python/pants_test/util/test_collections.py
new file mode 100644
index 00000000000..559a5e0ad00
--- /dev/null
+++ b/tests/python/pants_test/util/test_collections.py
@@ -0,0 +1,22 @@
+# coding=utf-8
+# Copyright 2017 Pants project contributors (see CONTRIBUTORS.md).
+# Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+from __future__ import (absolute_import, division, generators, nested_scopes, print_function,
+                        unicode_literals, with_statement)
+
+import unittest
+
+from pants.util.collections import combined_dict
+
+
+class TestCollections(unittest.TestCase):
+  def test_combined_dict(self):
+    self.assertEqual(
+      combined_dict(
+       {'a': 1, 'b': 1, 'c': 1},
+       {'b': 2, 'c': 2},
+       {'c': 3},
+      ),
+      {'a': 1, 'b': 2, 'c': 3}
+    )