diff --git a/build-support/bin/release.sh b/build-support/bin/release.sh
index f7e83c9fca1..281cf878e9e 100755
--- a/build-support/bin/release.sh
+++ b/build-support/bin/release.sh
@@ -155,6 +155,7 @@ function execute_packaged_pants_with_internal_backends() {
     --no-verify-config \
     --pythonpath="['pants-plugins/src/python']" \
     --backend-packages="[\
+        'pants.rules.core',\
         'pants.backend.codegen',\
         'pants.backend.docgen',\
         'pants.backend.graph_info',\
diff --git a/pants-plugins/src/python/internal_backend/rules_for_testing/register.py b/pants-plugins/src/python/internal_backend/rules_for_testing/register.py
index c25eb3150eb..c5728f94400 100644
--- a/pants-plugins/src/python/internal_backend/rules_for_testing/register.py
+++ b/pants-plugins/src/python/internal_backend/rules_for_testing/register.py
@@ -4,25 +4,26 @@
 
 from __future__ import absolute_import, division, print_function, unicode_literals
 
-from builtins import str
-
 from pants.engine.addressable import BuildFileAddresses
 from pants.engine.console import Console
+from pants.engine.goal import Goal
 from pants.engine.rules import console_rule
-from pants.engine.selectors import Get
-from pants.option.scope import Scope, ScopedOptions
-from pants.rules.core.exceptions import GracefulTerminationException
 
 
-@console_rule('list-and-die-for-testing', [Console, BuildFileAddresses])
-def fast_list_and_die_for_testing(console, addresses):
+class ListAndDieForTesting(Goal):
   """A fast and deadly variant of `./pants list`."""
-  options = yield Get(ScopedOptions, Scope(str('list')))
-  console.print_stderr('>>> Got an interesting option: {}'.format(options.options.documented))
+
+  name = 'list-and-die-for-testing'
+
+
+@console_rule(ListAndDieForTesting, [Console, BuildFileAddresses])
+def fast_list_and_die_for_testing(console, addresses):
   for address in addresses.dependencies:
     console.print_stdout(address.spec)
-  raise GracefulTerminationException(exit_code=42)
+  yield ListAndDieForTesting(exit_code=42)
 
 
 def rules():
-  return (fast_list_and_die_for_testing,)
+  return [
+      fast_list_and_die_for_testing,
+    ]
diff --git a/pants.ini b/pants.ini
index 18c53b66c33..61585b447f8 100644
--- a/pants.ini
+++ b/pants.ini
@@ -435,6 +435,5 @@ compiler_option_sets_disabled_args: {
 [libc]
 enable_libc_search: True
 
-[sourcefile_validation]
+[sourcefile-validation]
 config: @build-support/regexes/config.yaml
-detail_level: nonmatching
diff --git a/src/python/pants/backend/graph_info/register.py b/src/python/pants/backend/graph_info/register.py
index 9f991d505dd..e9e68d39a97 100644
--- a/src/python/pants/backend/graph_info/register.py
+++ b/src/python/pants/backend/graph_info/register.py
@@ -8,7 +8,6 @@
 from pants.backend.graph_info.tasks.dependees import ReverseDepmap
 from pants.backend.graph_info.tasks.filemap import Filemap
 from pants.backend.graph_info.tasks.filter import Filter
-from pants.backend.graph_info.tasks.list_targets import ListTargets
 from pants.backend.graph_info.tasks.minimal_cover import MinimalCover
 from pants.backend.graph_info.tasks.paths import Path, Paths
 from pants.backend.graph_info.tasks.sort_targets import SortTargets
@@ -16,7 +15,6 @@
 
 
 def register_goals():
-  task(name='list', action=ListTargets).install()
   task(name='path', action=Path).install()
   task(name='paths', action=Paths).install()
   task(name='dependees', action=ReverseDepmap).install()
diff --git a/src/python/pants/backend/graph_info/tasks/list_targets.py b/src/python/pants/backend/graph_info/tasks/list_targets.py
deleted file mode 100644
index 44a23c5fb73..00000000000
--- a/src/python/pants/backend/graph_info/tasks/list_targets.py
+++ /dev/null
@@ -1,76 +0,0 @@
-# coding=utf-8
-# Copyright 2014 Pants project contributors (see CONTRIBUTORS.md).
-# Licensed under the Apache License, Version 2.0 (see LICENSE).
-
-from __future__ import absolute_import, division, print_function, unicode_literals
-
-from builtins import str
-
-from pants.base.exceptions import TaskError
-from pants.task.console_task import ConsoleTask
-
-
-class ListTargets(ConsoleTask):
-  """Lists all targets matching the target specs.
-
-  If no targets are specified, lists all targets in the workspace.
-  """
-
-  @classmethod
-  def register_options(cls, register):
-    super(ListTargets, cls).register_options(register)
-    register('--provides', type=bool,
-             help='List only targets that provide an artifact, displaying the columns specified by '
-                  '--provides-columns.')
-    register('--provides-columns', default='address,artifact_id',
-             help='Display these columns when --provides is specified. Available columns are: '
-                  'address, artifact_id, repo_name, repo_url, push_db_basedir')
-    register('--documented', type=bool,
-             help='Print only targets that are documented with a description.')
-
-  def __init__(self, *args, **kwargs):
-    super(ListTargets, self).__init__(*args, **kwargs)
-    options = self.get_options()
-    self._provides = options.provides
-    self._provides_columns = options.provides_columns
-    self._documented = options.documented
-
-  def console_output(self, targets):
-    if self._provides:
-      extractors = dict(
-          address=lambda target: target.address.spec,
-          artifact_id=lambda target: str(target.provides),
-          repo_name=lambda target: target.provides.repo.name,
-          repo_url=lambda target: target.provides.repo.url,
-          push_db_basedir=lambda target: target.provides.repo.push_db_basedir,
-      )
-
-      def print_provides(column_extractors, target):
-        if getattr(target, 'provides', None):
-          return ' '.join(extractor(target) for extractor in column_extractors)
-
-      try:
-        column_extractors = [extractors[col] for col in (self._provides_columns.split(','))]
-      except KeyError:
-        raise TaskError('Invalid columns specified: {0}. Valid columns are: address, artifact_id, '
-                        'repo_name, repo_url, push_db_basedir.'.format(self._provides_columns))
-
-      print_fn = lambda target: print_provides(column_extractors, target)
-    elif self._documented:
-      def print_documented(target):
-        if target.description:
-          return '{0}\n  {1}'.format(target.address.spec,
-                                     '\n  '.join(target.description.strip().split('\n')))
-      print_fn = print_documented
-    else:
-      print_fn = lambda target: target.address.spec
-
-    visited = set()
-    for target in self.determine_target_roots('list'):
-      if target.is_synthetic:
-        continue
-      result = print_fn(target)
-      if result and result not in visited:
-        visited.add(result)
-
-    return sorted(visited)
diff --git a/src/python/pants/backend/project_info/rules/source_file_validator.py b/src/python/pants/backend/project_info/rules/source_file_validator.py
index 7a252f10237..62aec840e7d 100644
--- a/src/python/pants/backend/project_info/rules/source_file_validator.py
+++ b/src/python/pants/backend/project_info/rules/source_file_validator.py
@@ -9,13 +9,14 @@
 
 from future.utils import text_type
 
+from pants.base.exiter import PANTS_FAILED_EXIT_CODE, PANTS_SUCCEEDED_EXIT_CODE
 from pants.engine.console import Console
 from pants.engine.fs import Digest, FilesContent
+from pants.engine.goal import Goal
 from pants.engine.legacy.graph import HydratedTarget, HydratedTargets
 from pants.engine.objects import Collection
 from pants.engine.rules import console_rule, optionable_rule, rule
 from pants.engine.selectors import Get
-from pants.rules.core.exceptions import GracefulTerminationException
 from pants.subsystem.subsystem import Subsystem
 from pants.util.memo import memoized_method
 from pants.util.objects import datatype, enum
@@ -32,8 +33,18 @@ class DetailLevel(enum(['none', 'summary', 'nonmatching', 'all'])):
   pass
 
 
+class Validate(Goal):
+  name = 'validate'
+
+  @classmethod
+  def register_options(cls, register):
+    super(Validate, cls).register_options(register)
+    register('--detail-level', type=DetailLevel, default=DetailLevel.nonmatching,
+             help='How much detail to emit to the console.')
+
+
 class SourceFileValidation(Subsystem):
-  options_scope = 'sourcefile_validation'
+  options_scope = 'sourcefile-validation'
 
   @classmethod
   def register_options(cls, register):
@@ -68,8 +79,6 @@ def register_options(cls, register):
     register('--config', type=dict, fromfile=True,
              # TODO: Replace "See documentation" with actual URL, once we have some.
              help='Source file regex matching config.  See documentation for config schema.')
-    register('--detail-level', type=DetailLevel, default=DetailLevel.nonmatching,
-             help='How much detail to emit to the console.')
 
   @memoized_method
   def get_multi_matcher(self):
@@ -201,12 +210,12 @@ def get_applicable_content_pattern_names(self, path):
 
 # TODO: Switch this to `lint` once we figure out a good way for v1 tasks and v2 rules
 # to share goal names.
-@console_rule('validate', [Console, HydratedTargets, SourceFileValidation])
-def validate(console, hydrated_targets, source_file_validation):
+@console_rule(Validate, [Console, HydratedTargets, Validate.Options])
+def validate(console, hydrated_targets, validate_options):
   per_tgt_rmrs = yield [Get(RegexMatchResults, HydratedTarget, ht) for ht in hydrated_targets]
   regex_match_results = list(itertools.chain(*per_tgt_rmrs))
 
-  detail_level = source_file_validation.get_options().detail_level
+  detail_level = validate_options.values.detail_level
   regex_match_results = sorted(regex_match_results, key=lambda x: x.path)
   num_matched_all = 0
   num_nonmatched_some = 0
@@ -232,7 +241,11 @@ def validate(console, hydrated_targets, source_file_validation):
       num_nonmatched_some))
 
   if num_nonmatched_some:
-    raise GracefulTerminationException('Files failed validation.')
+    console.print_stderr('Files failed validation.')
+    exit_code = PANTS_FAILED_EXIT_CODE
+  else:
+    exit_code = PANTS_SUCCEEDED_EXIT_CODE
+  yield Validate(exit_code)
 
 
 @rule(RegexMatchResults, [HydratedTarget, SourceFileValidation])
diff --git a/src/python/pants/bin/daemon_pants_runner.py b/src/python/pants/bin/daemon_pants_runner.py
index fb88ae59345..fb5e9e1c542 100644
--- a/src/python/pants/bin/daemon_pants_runner.py
+++ b/src/python/pants/bin/daemon_pants_runner.py
@@ -17,13 +17,12 @@
 
 from pants.base.build_environment import get_buildroot
 from pants.base.exception_sink import ExceptionSink, SignalHandler
-from pants.base.exiter import PANTS_SUCCEEDED_EXIT_CODE, Exiter
+from pants.base.exiter import PANTS_FAILED_EXIT_CODE, PANTS_SUCCEEDED_EXIT_CODE, Exiter
 from pants.bin.local_pants_runner import LocalPantsRunner
 from pants.init.util import clean_global_runtime_state
 from pants.java.nailgun_io import NailgunStreamStdinReader, NailgunStreamWriter
 from pants.java.nailgun_protocol import ChunkType, NailgunProtocol
 from pants.pantsd.process_manager import ProcessManager
-from pants.rules.core.exceptions import GracefulTerminationException
 from pants.util.contextutil import hermetic_environment_as, stdio_as
 from pants.util.socket import teardown_socket
 
@@ -77,6 +76,31 @@ def exit(self, result=0, msg=None, *args, **kwargs):
       super(DaemonExiter, self).exit(result=result, *args, **kwargs)
 
 
+class _GracefulTerminationException(Exception):
+  """Allows for deferring the returning of the exit code of prefork work until post fork.
+
+  TODO: Once the fork boundary is removed in #7390, this class can be replaced by directly exiting
+  with the relevant exit code.
+  """
+
+  def __init__(self, exit_code=PANTS_FAILED_EXIT_CODE):
+    """
+    :param int exit_code: an optional exit code (defaults to PANTS_FAILED_EXIT_CODE)
+    """
+    super(_GracefulTerminationException, self).__init__('Terminated with {}'.format(exit_code))
+
+    if exit_code == PANTS_SUCCEEDED_EXIT_CODE:
+      raise ValueError(
+        "Cannot create _GracefulTerminationException with a successful exit code of {}"
+        .format(PANTS_SUCCEEDED_EXIT_CODE))
+
+    self._exit_code = exit_code
+
+  @property
+  def exit_code(self):
+    return self._exit_code
+
+
 class DaemonPantsRunner(ProcessManager):
   """A daemonizing PantsRunner that speaks the nailgun protocol to a remote client.
 
@@ -92,8 +116,8 @@ def create(cls, sock, args, env, services, scheduler_service):
       with cls.nailgunned_stdio(sock, env, handle_stdin=False):
         options, _, options_bootstrapper = LocalPantsRunner.parse_options(args, env)
         subprocess_dir = options.for_global_scope().pants_subprocessdir
-        graph_helper, target_roots = scheduler_service.prefork(options, options_bootstrapper)
-        deferred_exc = None
+        graph_helper, target_roots, exit_code = scheduler_service.prefork(options, options_bootstrapper)
+        deferred_exc = None if exit_code == PANTS_SUCCEEDED_EXIT_CODE else _GracefulTerminationException(exit_code)
     except Exception:
       deferred_exc = sys.exc_info()
       graph_helper = None
@@ -316,9 +340,6 @@ def post_fork_child(self):
         # Clean global state.
         clean_global_runtime_state(reset_subsystem=True)
 
-        # Re-raise any deferred exceptions, if present.
-        self._raise_deferred_exc()
-
         # Otherwise, conduct a normal run.
         runner = LocalPantsRunner.create(
           self._exiter,
@@ -329,10 +350,14 @@ def post_fork_child(self):
           self._options_bootstrapper
         )
         runner.set_start_time(self._maybe_get_client_start_time_from_env(self._env))
+
+        # Re-raise any deferred exceptions, if present.
+        self._raise_deferred_exc()
+
         runner.run()
       except KeyboardInterrupt:
         self._exiter.exit_and_fail('Interrupted by user.\n')
-      except GracefulTerminationException as e:
+      except _GracefulTerminationException as e:
         ExceptionSink.log_exception(
           'Encountered graceful termination exception {}; exiting'.format(e))
         self._exiter.exit(e.exit_code)
diff --git a/src/python/pants/bin/goal_runner.py b/src/python/pants/bin/goal_runner.py
index ed036f5afb5..887a779ff96 100644
--- a/src/python/pants/bin/goal_runner.py
+++ b/src/python/pants/bin/goal_runner.py
@@ -15,7 +15,6 @@
 from pants.goal.context import Context
 from pants.goal.goal import Goal
 from pants.goal.run_tracker import RunTracker
-from pants.help.help_printer import HelpPrinter
 from pants.java.nailgun_executor import NailgunProcessGroup
 from pants.option.ranked_value import RankedValue
 from pants.task.task import QuietTaskMixin
@@ -51,15 +50,11 @@ def __init__(self, root_dir, options, build_config, run_tracker, reporting, grap
     self._explain = self._global_options.explain
     self._kill_nailguns = self._global_options.kill_nailguns
 
-  def _maybe_handle_help(self, help_request):
-    """Handle requests for `help` information."""
-    if help_request:
-      help_printer = HelpPrinter(self._options)
-      result = help_printer.print_help()
-      self._exiter(result)
-
-  def _determine_goals(self, address_mapper, requested_goals):
+  def _determine_v1_goals(self, address_mapper, options):
     """Check and populate the requested goals for a given run."""
+    v1_goals, ambiguous_goals, _ = options.goals_by_version
+    requested_goals = v1_goals + ambiguous_goals
+
     spec_parser = CmdLineSpecParser(self._root_dir)
 
     for goal in requested_goals:
@@ -95,7 +90,7 @@ def _setup_context(self):
         self._root_dir
       )
 
-      goals = self._determine_goals(address_mapper, self._options.goals)
+      goals = self._determine_v1_goals(address_mapper, self._options)
       is_quiet = self._should_be_quiet(goals)
 
       target_root_instances = self._roots_to_targets(build_graph, self._target_roots)
@@ -122,7 +117,6 @@ def _setup_context(self):
       return goals, context
 
   def create(self):
-    self._maybe_handle_help(self._options.help_request)
     goals, context = self._setup_context()
     return GoalRunner(context=context,
                       goals=goals,
@@ -131,7 +125,11 @@ def create(self):
 
 
 class GoalRunner(object):
-  """Lists installed goals or else executes a named goal."""
+  """Lists installed goals or else executes a named goal.
+
+  NB: GoalRunner represents a v1-only codepath. v2 goals are registered via `@console_rule` and
+  the `pants.engine.goal.Goal` class.
+  """
 
   Factory = GoalRunnerFactory
 
diff --git a/src/python/pants/bin/local_pants_runner.py b/src/python/pants/bin/local_pants_runner.py
index 81b1f6223f1..16cd3c4a90c 100644
--- a/src/python/pants/bin/local_pants_runner.py
+++ b/src/python/pants/bin/local_pants_runner.py
@@ -5,7 +5,6 @@
 from __future__ import absolute_import, division, print_function, unicode_literals
 
 import logging
-import traceback
 from builtins import object, str
 
 from pants.base.build_environment import get_buildroot
@@ -15,6 +14,7 @@
 from pants.bin.goal_runner import GoalRunner
 from pants.engine.native import Native
 from pants.goal.run_tracker import RunTracker
+from pants.help.help_printer import HelpPrinter
 from pants.init.engine_initializer import EngineInitializer
 from pants.init.logging import setup_logging_from_options
 from pants.init.options_initializer import BuildConfigInitializer, OptionsInitializer
@@ -22,7 +22,6 @@
 from pants.init.target_roots_calculator import TargetRootsCalculator
 from pants.option.options_bootstrapper import OptionsBootstrapper
 from pants.reporting.reporting import Reporting
-from pants.rules.core.exceptions import GracefulTerminationException
 from pants.util.contextutil import maybe_profiled
 
 
@@ -235,8 +234,16 @@ def run(self):
     with maybe_profiled(self._profile_path):
       self._run()
 
+  def _maybe_handle_help(self):
+    """Handle requests for `help` information."""
+    if self._options.help_request:
+      help_printer = HelpPrinter(self._options)
+      result = help_printer.print_help()
+      self._exiter(result)
+
   def _maybe_run_v1(self):
-    if not self._global_options.v1:
+    v1_goals, ambiguous_goals, _ = self._options.goals_by_version
+    if not v1_goals and (not ambiguous_goals or not self._global_options.v1):
       return PANTS_SUCCEEDED_EXIT_CODE
 
     # Setup and run GoalRunner.
@@ -255,30 +262,20 @@ def _maybe_run_v1(self):
   def _maybe_run_v2(self):
     # N.B. For daemon runs, @console_rules are invoked pre-fork -
     # so this path only serves the non-daemon run mode.
-    if self._is_daemon or not self._global_options.v2:
+    if self._is_daemon:
       return PANTS_SUCCEEDED_EXIT_CODE
 
-    # If we're a pure --v2 run, validate goals - otherwise some goals specified
-    # may be provided by the --v1 task paths.
-    if not self._global_options.v1:
-      self._graph_session.validate_goals(self._options.goals_and_possible_v2_goals)
-
-    try:
-      self._graph_session.run_console_rules(
-        self._options_bootstrapper,
-        self._options.goals_and_possible_v2_goals,
-        self._target_roots,
-      )
-    except GracefulTerminationException as e:
-      logger.debug('Encountered graceful termination exception {}; exiting'.format(e))
-      return e.exit_code
-    except Exception:
-      logger.warning('Encountered unhandled exception during rule execution!')
-      logger.warning(traceback.format_exc())
-      return PANTS_FAILED_EXIT_CODE
-    else:
+    _, ambiguous_goals, v2_goals = self._options.goals_by_version
+    goals = v2_goals + (ambiguous_goals if self._global_options.v2 else tuple())
+    if not goals:
       return PANTS_SUCCEEDED_EXIT_CODE
 
+    return self._graph_session.run_console_rules(
+      self._options_bootstrapper,
+      goals,
+      self._target_roots,
+    )
+
   @staticmethod
   def _compute_final_exit_code(*codes):
     """Returns the exit code with higher abs value in case of negative values."""
@@ -290,6 +287,8 @@ def _compute_final_exit_code(*codes):
 
   def _run(self):
     try:
+      self._maybe_handle_help()
+
       engine_result = self._maybe_run_v2()
       goal_runner_result = self._maybe_run_v1()
     finally:
diff --git a/src/python/pants/core_tasks/register.py b/src/python/pants/core_tasks/register.py
index 3ee5dade182..b1144caa1d9 100644
--- a/src/python/pants/core_tasks/register.py
+++ b/src/python/pants/core_tasks/register.py
@@ -37,7 +37,6 @@ def register_goals():
   Goal.register('binary', 'Create a runnable binary.')
   Goal.register('resources', 'Prepare resources.')
   Goal.register('bundle', 'Create a deployable application bundle.')
-  Goal.register('test', 'Run tests.')
   Goal.register('bench', 'Run benchmarks.')
   Goal.register('repl', 'Run a REPL.')
   Goal.register('repl-dirty', 'Run a REPL, skipping compilation.')
@@ -87,7 +86,7 @@ def register_goals():
   task(name='compile-prep-command', action=RunCompilePrepCommand).install('compile', first=True)
 
   # Stub for other goals to schedule 'test'. See noop_exec_task.py for why this is useful.
-  task(name='test', action=NoopTest).install('test')
+  task(name='legacy', action=NoopTest).install('test')
 
   # Workspace information.
   task(name='roots', action=ListRoots).install()
diff --git a/src/python/pants/engine/BUILD b/src/python/pants/engine/BUILD
index 7cb4ec7efd9..997b5526509 100644
--- a/src/python/pants/engine/BUILD
+++ b/src/python/pants/engine/BUILD
@@ -60,6 +60,15 @@ python_library(
   ]
 )
 
+python_library(
+  name='goal',
+  sources=['goal.py'],
+  dependencies=[
+    'src/python/pants/option',
+    'src/python/pants/util:meta',
+  ]
+)
+
 python_library(
   name='build_files',
   sources=['build_files.py'],
@@ -147,6 +156,7 @@ python_library(
     '3rdparty/python:asttokens',
     '3rdparty/python:future',
     '3rdparty/python/twitter/commons:twitter.common.collections',
+    ':goal',
     ':selectors',
     'src/python/pants/base:specs',
     'src/python/pants/util:collections',
diff --git a/src/python/pants/engine/console.py b/src/python/pants/engine/console.py
index 5e6708738df..aa48e121d1e 100644
--- a/src/python/pants/engine/console.py
+++ b/src/python/pants/engine/console.py
@@ -8,13 +8,17 @@
 
 
 class Console(object):
+  def __init__(self, stdout=None, stderr=None):
+    self._stdout = stdout or sys.stdout
+    self._stderr = stderr or sys.stderr
+
   @property
   def stdout(self):
-    return sys.stdout
+    return self._stdout
 
   @property
   def stderr(self):
-    return sys.stderr
+    return self._stderr
 
   def write_stdout(self, payload):
     self.stdout.write(payload)
diff --git a/src/python/pants/engine/goal.py b/src/python/pants/engine/goal.py
new file mode 100644
index 00000000000..7ad673591b6
--- /dev/null
+++ b/src/python/pants/engine/goal.py
@@ -0,0 +1,151 @@
+# coding=utf-8
+# Copyright 2019 Pants project contributors (see CONTRIBUTORS.md).
+# Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+from __future__ import absolute_import, division, print_function, unicode_literals
+
+from abc import abstractmethod
+from contextlib import contextmanager
+
+from pants.cache.cache_setup import CacheSetup
+from pants.option.optionable import Optionable
+from pants.option.scope import ScopeInfo
+from pants.subsystem.subsystem_client_mixin import SubsystemClientMixin
+from pants.util.memo import memoized_classproperty
+from pants.util.meta import AbstractClass, classproperty
+from pants.util.objects import datatype
+
+
+class Goal(datatype([('exit_code', int)]), AbstractClass):
+  """The named product of a `@console_rule`.
+  
+  This abstract class should be subclassed and given a `Goal.name` that it will be referred to by
+  when invoked from the command line. The `Goal.name` also acts as the options_scope for the `Goal`.
+
+  Since `@console_rules` always run in order to produce sideeffects (generally: console output), they
+  are not cacheable, and the `Goal` product of a `@console_rule` contains only a exit_code value to
+  indicate whether the rule exited cleanly.
+
+  Options values for a Goal can be retrived by declaring a dependency on the relevant `Goal.Options`
+  class.
+  """
+
+  @classproperty
+  @abstractmethod
+  def name(cls):
+    """The name used to select this Goal on the commandline, and for its options."""
+
+  @classproperty
+  def deprecated_cache_setup_removal_version(cls):
+    """Optionally defines a deprecation version for a CacheSetup dependency.
+
+    If this Goal should have an associated deprecated instance of `CacheSetup` (which was implicitly
+    required by all v1 Tasks), subclasses may set this to a valid deprecation version to create
+    that association.
+    """
+    return None
+
+  @classmethod
+  def register_options(cls, register):
+    """Register options for the `Goal.Options` of this `Goal`.
+
+    Subclasses may override and call register(*args, **kwargs). Callers can retrieve the resulting
+    options values by declaring a dependency on the `Goal.Options` class.
+    """
+
+  @memoized_classproperty
+  def Options(cls):
+    # NB: The naming of this property is terribly evil. But this construction allows the inner class
+    # to get a reference to the outer class, which avoids implementers needing to subclass the inner
+    # class in order to define their options values, while still allowing for the useful namespacing
+    # of `Goal.Options`.
+    outer_cls = cls
+    class _Options(SubsystemClientMixin, Optionable, _GoalOptions):
+      @classproperty
+      def options_scope(cls):
+        return outer_cls.name
+
+      @classmethod
+      def register_options(cls, register):
+        super(_Options, cls).register_options(register)
+        # Delegate to the outer class.
+        outer_cls.register_options(register)
+
+      @classmethod
+      def subsystem_dependencies(cls):
+        # NB: `Goal.Options` implements `SubsystemClientMixin` in order to allow v1 `Tasks` to
+        # depend on v2 Goals, and for `Goals` to declare a deprecated dependency on a `CacheSetup`
+        # instance for backwards compatibility purposes. But v2 Goals should _not_ have subsystem
+        # dependencies: instead, the @rules participating (transitively) in a Goal should directly
+        # declare their Subsystem deps.
+        if outer_cls.deprecated_cache_setup_removal_version:
+          dep = CacheSetup.scoped(
+              cls,
+              removal_version=outer_cls.deprecated_cache_setup_removal_version,
+              removal_hint='Goal `{}` uses an independent caching implementation, and ignores `{}`.'.format(
+                cls.options_scope,
+                CacheSetup.subscope(cls.options_scope),
+              )
+            )
+          return (dep,)
+        return tuple()
+
+      options_scope_category = ScopeInfo.GOAL
+
+      def __init__(self, scope, scoped_options):
+        # NB: This constructor is shaped to meet the contract of `Optionable(Factory).signature`.
+        super(_Options, self).__init__()
+        self._scope = scope
+        self._scoped_options = scoped_options
+
+      @property
+      def values(self):
+        """Returns the option values for these Goal.Options."""
+        return self._scoped_options
+    _Options.__doc__ = outer_cls.__doc__
+    return _Options
+
+
+class _GoalOptions(object):
+  """A marker trait for the anonymous inner `Goal.Options` classes for `Goal`s."""
+
+
+class LineOriented(object):
+  """A mixin for Goal that adds Options to support the `line_oriented` context manager."""
+
+  @classmethod
+  def register_options(cls, register):
+    super(LineOriented, cls).register_options(register)
+    register('--sep', default='\\n', metavar='<separator>',
+             help='String to use to separate result lines.')
+    register('--output-file', metavar='<path>',
+             help='Write line-oriented output to this file instead.')
+
+  @classmethod
+  @contextmanager
+  def line_oriented(cls, line_oriented_options, console):
+    """Given Goal.Options and a Console, yields functions for writing to stdout and stderr, respectively.
+
+    The passed options instance will generally be the `Goal.Options` of a `LineOriented` `Goal`.
+    """
+    if type(line_oriented_options) != cls.Options:
+      raise AssertionError(
+          'Expected Options for `{}`, got: {}'.format(cls.__name__, line_oriented_options))
+
+    output_file = line_oriented_options.values.output_file
+    sep = line_oriented_options.values.sep.encode('utf-8').decode('unicode_escape')
+
+    stdout, stderr = console.stdout, console.stderr
+    if output_file:
+      stdout = open(output_file, 'w')
+
+    try:
+      print_stdout = lambda msg: print(msg, file=stdout, end=sep)
+      print_stderr = lambda msg: print(msg, file=stderr)
+      yield print_stdout, print_stderr
+    finally:
+      if output_file:
+        stdout.close()
+      else:
+        stdout.flush()
+      stderr.flush()
diff --git a/src/python/pants/engine/legacy/graph.py b/src/python/pants/engine/legacy/graph.py
index 0720b600431..de8d56f3f47 100644
--- a/src/python/pants/engine/legacy/graph.py
+++ b/src/python/pants/engine/legacy/graph.py
@@ -204,8 +204,7 @@ def inject_addresses_closure(self, addresses):
     if not addresses:
       return
     dependencies = tuple(SingleAddress(a.spec_path, a.target_name) for a in addresses)
-    specs = [Specs(dependencies=tuple(dependencies))]
-    for _ in self._inject_specs(specs):
+    for _ in self._inject_specs(Specs(dependencies=tuple(dependencies))):
       pass
 
   def inject_roots_closure(self, target_roots, fail_fast=None):
@@ -213,9 +212,8 @@ def inject_roots_closure(self, target_roots, fail_fast=None):
       yield address
 
   def inject_specs_closure(self, specs, fail_fast=None):
-    specs = [Specs(dependencies=tuple(specs))]
     # Request loading of these specs.
-    for address in self._inject_specs(specs):
+    for address in self._inject_specs(Specs(dependencies=tuple(specs))):
       yield address
 
   def resolve_address(self, address):
@@ -251,18 +249,18 @@ def _inject_addresses(self, subjects):
         yielded_addresses.add(address)
         yield address
 
-  def _inject_specs(self, subjects):
-    """Injects targets into the graph for each of the given `Spec` objects.
+  def _inject_specs(self, specs):
+    """Injects targets into the graph for the given `Specs` object.
 
     Yields the resulting addresses.
     """
-    if not subjects:
+    if not specs:
       return
 
-    logger.debug('Injecting specs to %s: %s', self, subjects)
+    logger.debug('Injecting specs to %s: %s', self, specs)
     with self._resolve_context():
       thts, = self._scheduler.product_request(TransitiveHydratedTargets,
-                                              subjects)
+                                              [specs])
 
     self._index(thts.closure)
 
diff --git a/src/python/pants/engine/rules.py b/src/python/pants/engine/rules.py
index 72194a35342..413e6f75345 100644
--- a/src/python/pants/engine/rules.py
+++ b/src/python/pants/engine/rules.py
@@ -5,19 +5,16 @@
 from __future__ import absolute_import, division, print_function, unicode_literals
 
 import ast
-import functools
 import inspect
 import itertools
 import logging
 import sys
 from abc import abstractproperty
-from builtins import bytes, str
-from types import GeneratorType
 
 import asttokens
-from future.utils import PY2
 from twitter.common.collections import OrderedSet
 
+from pants.engine.goal import Goal
 from pants.engine.selectors import Get
 from pants.util.collections import assert_single_element
 from pants.util.collections_abc_backport import Iterable, OrderedDict
@@ -174,42 +171,6 @@ def visit_Yield(self, node):
 """))
 
 
-class _GoalProduct(object):
-  """GoalProduct is a factory for anonymous singleton types representing the execution of goals.
-
-  The created types are returned by `@console_rule` instances, which may not have any outputs
-  of their own.
-  """
-  PRODUCT_MAP = {}
-
-  @staticmethod
-  def _synthesize_goal_product(name):
-    product_type_name = '{}GoalExecution'.format(name.capitalize())
-    if PY2:
-      product_type_name = product_type_name.encode('utf-8')
-    return type(product_type_name, (datatype([]),), {})
-
-  @classmethod
-  def for_name(cls, name):
-    assert isinstance(name, (bytes, str))
-    if name is bytes:
-      name = name.decode('utf-8')
-    if name not in cls.PRODUCT_MAP:
-      cls.PRODUCT_MAP[name] = cls._synthesize_goal_product(name)
-    return cls.PRODUCT_MAP[name]
-
-
-def _terminated(generator, terminator):
-  """A generator that "appends" the given terminator value to the given generator."""
-  gen_input = None
-  try:
-    while True:
-      res = generator.send(gen_input)
-      gen_input = yield res
-  except StopIteration:
-    yield terminator
-
-
 @memoized
 def optionable_rule(optionable_factory):
   """Returns a TaskRule that constructs an instance of the Optionable for the given OptionableFactory.
@@ -230,15 +191,22 @@ def _get_starting_indent(source):
   return 0
 
 
-def _make_rule(output_type, input_selectors, for_goal=None, cacheable=True):
+def _make_rule(output_type, input_selectors, cacheable=True):
   """A @decorator that declares that a particular static function may be used as a TaskRule.
 
+  As a special case, if the output_type is a subclass of `Goal`, the `Goal.Options` for the `Goal`
+  are registered as dependency Optionables.
+
   :param type output_type: The return/output type for the Rule. This must be a concrete Python type.
   :param list input_selectors: A list of Selector instances that matches the number of arguments
     to the @decorated function.
-  :param str for_goal: If this is a @console_rule, which goal string it's called for.
   """
 
+  is_goal_cls = isinstance(output_type, type) and issubclass(output_type, Goal)
+  if is_goal_cls == cacheable:
+    raise TypeError('An `@rule` that produces a `Goal` must be declared with @console_rule in order '
+                    'to signal that it is not cacheable.')
+
   def wrapper(func):
     if not inspect.isfunction(func):
       raise ValueError('The @rule decorator must be applied innermost of all decorators.')
@@ -282,31 +250,22 @@ def resolve_type(name):
       Get.create_statically_for_rule_graph(resolve_type(p), resolve_type(s))
       for p, s in rule_visitor.gets)
 
-    # For @console_rule, redefine the function to avoid needing a literal return of the output type.
-    if for_goal:
-      def goal_and_return(*args, **kwargs):
-        res = func(*args, **kwargs)
-        if isinstance(res, GeneratorType):
-          # Return a generator with an output_type instance appended.
-          return _terminated(res, output_type())
-        elif res is not None:
-          raise Exception('A @console_rule should not have a return value.')
-        return output_type()
-      functools.update_wrapper(goal_and_return, func)
-      wrapped_func = goal_and_return
+    # Register dependencies for @console_rule/Goal.
+    if is_goal_cls:
+      dependency_rules = (optionable_rule(output_type.Options),)
     else:
-      wrapped_func = func
+      dependency_rules = None
 
-    wrapped_func.rule = TaskRule(
+    func.rule = TaskRule(
         output_type,
         tuple(input_selectors),
-        wrapped_func,
+        func,
         input_gets=tuple(gets),
-        goal=for_goal,
+        dependency_rules=dependency_rules,
         cacheable=cacheable,
       )
 
-    return wrapped_func
+    return func
   return wrapper
 
 
@@ -314,9 +273,8 @@ def rule(output_type, input_selectors):
   return _make_rule(output_type, input_selectors)
 
 
-def console_rule(goal_name, input_selectors):
-  output_type = _GoalProduct.for_name(goal_name)
-  return _make_rule(output_type, input_selectors, goal_name, False)
+def console_rule(goal_cls, input_selectors):
+  return _make_rule(goal_cls, input_selectors, False)
 
 
 def union(cls):
@@ -374,7 +332,15 @@ class Rule(AbstractClass):
   def output_type(self):
     """An output `type` for the rule."""
 
-  @property
+  @abstractproperty
+  def dependency_rules(self):
+    """A tuple of @rules that are known to be necessary to run this rule.
+
+    Note that installing @rules as flat lists is generally preferable, as Rules already implicitly
+    form a loosely coupled RuleGraph: this facility exists only to assist with boilerplate removal.
+    """
+
+  @abstractproperty
   def dependency_optionables(self):
     """A tuple of Optionable classes that are known to be necessary to run this rule."""
     return ()
@@ -385,7 +351,7 @@ class TaskRule(datatype([
   ('input_selectors', TypedCollection(SubclassesOf(type))),
   ('input_gets', tuple),
   'func',
-  'goal',
+  ('dependency_rules', tuple),
   ('dependency_optionables', tuple),
   ('cacheable', bool),
 ]), Rule):
@@ -401,17 +367,18 @@ def __new__(cls,
               input_selectors,
               func,
               input_gets,
-              goal=None,
               dependency_optionables=None,
+              dependency_rules=None,
               cacheable=True):
 
+    # Create.
     return super(TaskRule, cls).__new__(
         cls,
         output_type,
         input_selectors,
         input_gets,
         func,
-        goal,
+        dependency_rules or tuple(),
         dependency_optionables or tuple(),
         cacheable,
       )
@@ -433,6 +400,14 @@ class RootRule(datatype([('output_type', _type_field)]), Rule):
   of an execution.
   """
 
+  @property
+  def dependency_rules(self):
+    return tuple()
+
+  @property
+  def dependency_optionables(self):
+    return tuple()
+
 
 # TODO: add typechecking here -- would need to have a TypedCollection for dicts for `union_rules`.
 class RuleIndex(datatype(['rules', 'roots', 'union_rules'])):
@@ -459,6 +434,8 @@ def add_rule(rule):
         add_root_rule(rule)
       else:
         add_task(rule.output_type, rule)
+      for dep_rule in rule.dependency_rules:
+        add_rule(dep_rule)
 
     def add_type_transition_rule(union_rule):
       # NB: This does not require that union bases be supplied to `def rules():`, as the union type
diff --git a/src/python/pants/engine/scheduler.py b/src/python/pants/engine/scheduler.py
index b5dfd0c3ba9..843bd202405 100644
--- a/src/python/pants/engine/scheduler.py
+++ b/src/python/pants/engine/scheduler.py
@@ -11,6 +11,7 @@
 from builtins import object, open, str, zip
 from types import GeneratorType
 
+from pants.base.exiter import PANTS_FAILED_EXIT_CODE
 from pants.base.project_tree import Dir, File, Link
 from pants.build_graph.address import Address
 from pants.engine.fs import (Digest, DirectoryToMaterialize, FileContent, FilesContent,
@@ -21,7 +22,6 @@
 from pants.engine.objects import Collection
 from pants.engine.rules import RuleIndex, TaskRule
 from pants.engine.selectors import Params
-from pants.rules.core.exceptions import GracefulTerminationException
 from pants.util.contextutil import temporary_file_path
 from pants.util.dirutil import check_no_overlapping_paths
 from pants.util.objects import datatype
@@ -484,8 +484,9 @@ def _trace_on_error(self, unique_exceptions, request):
 
   def run_console_rule(self, product, subject):
     """
-    :param product: product type for the request.
+    :param product: A Goal subtype.
     :param subject: subject for the request.
+    :returns: An exit_code for the given Goal.
     """
     request = self.execution_request([product], [subject])
     returns, throws = self.execute(request)
@@ -493,9 +494,10 @@ def run_console_rule(self, product, subject):
     if throws:
       _, state = throws[0]
       exc = state.exc
-      if isinstance(exc, GracefulTerminationException):
-        raise exc
       self._trace_on_error([exc], request)
+      return PANTS_FAILED_EXIT_CODE
+    _, state = returns[0]
+    return state.value.exit_code
 
   def product_request(self, product, subjects):
     """Executes a request for a single product for some subjects, and returns the products.
diff --git a/src/python/pants/help/help_printer.py b/src/python/pants/help/help_printer.py
index d0bbc05a511..41cdee1fd85 100644
--- a/src/python/pants/help/help_printer.py
+++ b/src/python/pants/help/help_printer.py
@@ -54,11 +54,9 @@ def _print_goals_help(self):
     print('\nUse `pants help $goal` to get help for a particular goal.\n')
     goal_descriptions = {}
     for scope_info in self._options.known_scope_to_info.values():
-      if scope_info.category not in (ScopeInfo.GOAL,):
+      if scope_info.category not in (ScopeInfo.GOAL, ScopeInfo.GOAL_V1):
         continue
-      components = scope_info.scope.split('.', 1)
-      if len(components) == 1 and scope_info.description:
-        goal_descriptions[scope_info.scope] = scope_info.description
+      goal_descriptions[scope_info.scope] = scope_info.description
     goal_descriptions.update({goal.name: goal.description_first_line
                               for goal in Goal.all()
                               if goal.description})
diff --git a/src/python/pants/init/engine_initializer.py b/src/python/pants/init/engine_initializer.py
index 86a21c608de..3f90ad4c7c7 100644
--- a/src/python/pants/init/engine_initializer.py
+++ b/src/python/pants/init/engine_initializer.py
@@ -16,12 +16,14 @@
 from pants.backend.python.targets.python_library import PythonLibrary
 from pants.backend.python.targets.python_tests import PythonTests
 from pants.base.build_environment import get_buildroot
+from pants.base.exiter import PANTS_SUCCEEDED_EXIT_CODE
 from pants.base.file_system_project_tree import FileSystemProjectTree
 from pants.build_graph.build_configuration import BuildConfiguration
 from pants.build_graph.remote_sources import RemoteSources
 from pants.engine.build_files import create_graph_rules
 from pants.engine.console import Console
 from pants.engine.fs import create_fs_rules
+from pants.engine.goal import Goal
 from pants.engine.isolated_process import create_process_rules
 from pants.engine.legacy.address_mapper import LegacyAddressMapper
 from pants.engine.legacy.graph import (LegacyBuildGraph, TransitiveHydratedTargets,
@@ -154,14 +156,6 @@ def __init__(self, invalid_goals):
       )
       self.invalid_goals = invalid_goals
 
-  @staticmethod
-  def _determine_subjects(target_roots):
-    """A utility to determines the subjects for the request.
-
-    :param TargetRoots target_roots: The targets root of the request.
-    """
-    return target_roots.specs or []
-
   def warm_product_graph(self, target_roots):
     """Warm the scheduler's `ProductGraph` with `TransitiveHydratedTargets` products.
 
@@ -171,19 +165,10 @@ def warm_product_graph(self, target_roots):
     :param TargetRoots target_roots: The targets root of the request.
     """
     logger.debug('warming target_roots for: %r', target_roots)
-    subjects = self._determine_subjects(target_roots)
+    subjects = [target_roots.specs]
     request = self.scheduler_session.execution_request([TransitiveHydratedTargets], subjects)
     self.scheduler_session.execute(request)
 
-  def validate_goals(self, goals):
-    """Checks for @console_rules that satisfy requested goals.
-
-    :param list goals: The list of requested goal names as passed on the commandline.
-    """
-    invalid_goals = [goal for goal in goals if goal not in self.goal_map]
-    if invalid_goals:
-      raise self.InvalidGoals(invalid_goals)
-
   def run_console_rules(self, options_bootstrapper, goals, target_roots):
     """Runs @console_rules sequentially and interactively by requesting their implicit Goal products.
 
@@ -191,23 +176,24 @@ def run_console_rules(self, options_bootstrapper, goals, target_roots):
 
     :param list goals: The list of requested goal names as passed on the commandline.
     :param TargetRoots target_roots: The targets root of the request.
+
+    :returns: An exit code.
     """
-    # Reduce to only applicable goals - with validation happening by way of `validate_goals()`.
-    goals = [goal for goal in goals if goal in self.goal_map]
-    subjects = self._determine_subjects(target_roots)
+    subject = target_roots.specs
     console = Console()
-    # Console rule can only have one subject.
-    # TODO: What about console_rules with no subjects (i.e., no target specs)?
-    assert len(subjects) == 1
     for goal in goals:
+      goal_product = self.goal_map[goal]
+      params = Params(subject, options_bootstrapper, console)
+      logger.debug('requesting {} to satisfy execution of `{}` goal'.format(goal_product, goal))
       try:
-        goal_product = self.goal_map[goal]
-        params = Params(subjects[0], options_bootstrapper, console)
-        logger.debug('requesting {} to satisfy execution of `{}` goal'.format(goal_product, goal))
-        self.scheduler_session.run_console_rule(goal_product, params)
+        exit_code = self.scheduler_session.run_console_rule(goal_product, params)
       finally:
         console.flush()
 
+      if exit_code != PANTS_SUCCEEDED_EXIT_CODE:
+        return exit_code
+    return PANTS_SUCCEEDED_EXIT_CODE
+
   def create_build_graph(self, target_roots, build_root=None):
     """Construct and return a `BuildGraph` given a set of input specs.
 
@@ -236,8 +222,11 @@ class GoalMappingError(Exception):
   @staticmethod
   def _make_goal_map_from_rules(rules):
     goal_map = {}
-    goal_to_rule = [(rule.goal, rule) for rule in rules if getattr(rule, 'goal', None) is not None]
-    for goal, r in goal_to_rule:
+    for r in rules:
+      output_type = getattr(r, 'output_type', None)
+      if not output_type or not issubclass(output_type, Goal):
+        continue
+      goal = r.output_type.name
       if goal in goal_map:
         raise EngineInitializer.GoalMappingError(
           'could not map goal `{}` to rule `{}`: already claimed by product `{}`'
diff --git a/src/python/pants/init/target_roots_calculator.py b/src/python/pants/init/target_roots_calculator.py
index 3479a11ed82..508d43a58e7 100644
--- a/src/python/pants/init/target_roots_calculator.py
+++ b/src/python/pants/init/target_roots_calculator.py
@@ -35,19 +35,16 @@ def parse_specs(cls, target_specs, build_root=None, exclude_patterns=None, tags=
 
     :param iterable target_specs: An iterable of string specs.
     :param string build_root: The path to the build root.
-    :returns: An `OrderedSet` of `Spec` objects.
+    :returns: A `Specs` object.
     """
     build_root = build_root or get_buildroot()
     spec_parser = CmdLineSpecParser(build_root)
 
     dependencies = tuple(OrderedSet(spec_parser.parse_spec(spec_str) for spec_str in target_specs))
-    if not dependencies:
-      return None
-    return [Specs(
+    return Specs(
       dependencies=dependencies,
       exclude_patterns=exclude_patterns if exclude_patterns else tuple(),
       tags=tags)
-    ]
 
   @classmethod
   def changed_files(cls, scm, changes_since=None, diffspec=None):
@@ -85,7 +82,7 @@ def create(cls, options, session, build_root=None, exclude_patterns=None, tags=N
     logger.debug('changed_request is: %s', changed_request)
     logger.debug('owned_files are: %s', owned_files)
     targets_specified = sum(1 for item
-                         in (changed_request.is_actionable(), owned_files, spec_roots)
+                         in (changed_request.is_actionable(), owned_files, spec_roots.dependencies)
                          if item)
 
     if targets_specified > 1:
@@ -112,7 +109,7 @@ def create(cls, options, session, build_root=None, exclude_patterns=None, tags=N
       changed_addresses, = session.product_request(BuildFileAddresses, [request])
       logger.debug('changed addresses: %s', changed_addresses)
       dependencies = tuple(SingleAddress(a.spec_path, a.target_name) for a in changed_addresses)
-      return TargetRoots([Specs(dependencies=dependencies, exclude_patterns=exclude_patterns, tags=tags)])
+      return TargetRoots(Specs(dependencies=dependencies, exclude_patterns=exclude_patterns, tags=tags))
 
     if owned_files:
       # We've been provided no spec roots (e.g. `./pants list`) AND a owner request. Compute
@@ -121,6 +118,6 @@ def create(cls, options, session, build_root=None, exclude_patterns=None, tags=N
       owner_addresses, = session.product_request(BuildFileAddresses, [request])
       logger.debug('owner addresses: %s', owner_addresses)
       dependencies = tuple(SingleAddress(a.spec_path, a.target_name) for a in owner_addresses)
-      return TargetRoots([Specs(dependencies=dependencies, exclude_patterns=exclude_patterns, tags=tags)])
+      return TargetRoots(Specs(dependencies=dependencies, exclude_patterns=exclude_patterns, tags=tags))
 
     return TargetRoots(spec_roots)
diff --git a/src/python/pants/option/options.py b/src/python/pants/option/options.py
index e974cd09695..5f4654e04c1 100644
--- a/src/python/pants/option/options.py
+++ b/src/python/pants/option/options.py
@@ -9,8 +9,6 @@
 import sys
 from builtins import object, open, str
 
-from twitter.common.collections import OrderedSet
-
 from pants.base.deprecated import warn_or_error
 from pants.option.arg_splitter import GLOBAL_SCOPE, ArgSplitter
 from pants.option.global_options import GlobalOptionsRegistrar
@@ -19,7 +17,7 @@
 from pants.option.option_value_container import OptionValueContainer
 from pants.option.parser_hierarchy import ParserHierarchy, all_enclosing_scopes, enclosing_scope
 from pants.option.scope import ScopeInfo
-from pants.util.memo import memoized_method
+from pants.util.memo import memoized_method, memoized_property
 
 
 def make_flag_regex(long_name, short_name=None):
@@ -81,6 +79,9 @@ class Options(object):
   class FrozenOptionsError(Exception):
     """Options are frozen and can't be mutated."""
 
+  class DuplicateScopeError(Exception):
+    """More than one registration occurred for the same scope."""
+
   @classmethod
   def complete_scopes(cls, scope_infos):
     """Expand a set of scopes to include all enclosing scopes.
@@ -90,13 +91,17 @@ def complete_scopes(cls, scope_infos):
     Also adds any deprecated scopes.
     """
     ret = {GlobalOptionsRegistrar.get_scope_info()}
-    original_scopes = set()
+    original_scopes = dict()
     for si in scope_infos:
       ret.add(si)
-      original_scopes.add(si.scope)
+      if si.scope in original_scopes:
+        raise cls.DuplicateScopeError('Scope `{}` claimed by {}, was also claimed by {}.'.format(
+            si.scope, si, original_scopes[si.scope]
+          ))
+      original_scopes[si.scope] = si
       if si.deprecated_scope:
         ret.add(ScopeInfo(si.deprecated_scope, si.category, si.optionable_cls))
-        original_scopes.add(si.deprecated_scope)
+        original_scopes[si.deprecated_scope] = si
 
     # TODO: Once scope name validation is enforced (so there can be no dots in scope name
     # components) we can replace this line with `for si in scope_infos:`, because it will
@@ -196,20 +201,28 @@ def goals(self):
     """
     return self._goals
 
-  # TODO: Replace this with a formal way of registering v2-only goals.
-  # See https://github.com/pantsbuild/pants/issues/6651
-  @property
-  def goals_and_possible_v2_goals(self):
-    """Goals, including any unrecognised scopes which may be v2-only goals.
+  @memoized_property
+  def goals_by_version(self):
+    """Goals organized into three tuples by whether they are v1, ambiguous, or v2 goals (respectively).
 
-    Experimental API which shouldn't be relied on outside of Pants itself.
+    It's possible for a goal to be implemented with both v1 and v2, in which case a consumer
+    should use the `--v1` and `--v2` global flags to disambiguate.
     """
-    if self._unknown_scopes:
-      r = OrderedSet(self.goals)
-      r.update(self._unknown_scopes)
-      return r
-    else:
-      return self.goals
+    v1, ambiguous, v2 = [], [], []
+    for goal in self._goals:
+      goal_dot = '{}.'.format(goal)
+      scope_categories = {s.category
+                          for s in self.known_scope_to_info.values()
+                          if s.scope == goal or s.scope.startswith(goal_dot)}
+      is_v1 = ScopeInfo.TASK in scope_categories
+      is_v2 = ScopeInfo.GOAL in scope_categories
+      if is_v1 and is_v2:
+        ambiguous.append(goal)
+      elif is_v1:
+        v1.append(goal)
+      else:
+        v2.append(goal)
+    return tuple(v1), tuple(ambiguous), tuple(v2)
 
   @property
   def known_scope_to_info(self):
diff --git a/src/python/pants/option/scope.py b/src/python/pants/option/scope.py
index 727f554840e..d6dbb9384ee 100644
--- a/src/python/pants/option/scope.py
+++ b/src/python/pants/option/scope.py
@@ -34,6 +34,7 @@ class ScopeInfo(datatype([
   # Symbolic constants for different categories of scope.
   GLOBAL = 'GLOBAL'
   GOAL = 'GOAL'
+  GOAL_V1 = 'GOAL_V1'
   TASK = 'TASK'
   SUBSYSTEM = 'SUBSYSTEM'
   INTERMEDIATE = 'INTERMEDIATE'  # Scope added automatically to fill out the scope hierarchy.
diff --git a/src/python/pants/pantsd/service/scheduler_service.py b/src/python/pants/pantsd/service/scheduler_service.py
index 9bdc4a491bd..ff02e2ab822 100644
--- a/src/python/pants/pantsd/service/scheduler_service.py
+++ b/src/python/pants/pantsd/service/scheduler_service.py
@@ -13,6 +13,7 @@
 
 from future.utils import PY3
 
+from pants.base.exiter import PANTS_SUCCEEDED_EXIT_CODE
 from pants.engine.fs import PathGlobs, Snapshot
 from pants.init.target_roots_calculator import TargetRootsCalculator
 from pants.pantsd.service.pants_service import PantsService
@@ -170,7 +171,7 @@ def product_graph_len(self):
   def prefork(self, options, options_bootstrapper):
     """Runs all pre-fork logic in the process context of the daemon.
 
-    :returns: `(LegacyGraphSession, TargetRoots)`
+    :returns: `(LegacyGraphSession, TargetRoots, exit_code)`
     """
     # If any nodes exist in the product graph, wait for the initial watchman event to avoid
     # racing watchman startup vs invalidation events.
@@ -181,18 +182,23 @@ def prefork(self, options, options_bootstrapper):
     v2_ui = options.for_global_scope().v2_ui
     zipkin_trace_v2 = options.for_scope('reporting').zipkin_trace_v2
     session = self._graph_helper.new_session(zipkin_trace_v2, v2_ui)
+
     if options.for_global_scope().loop:
-      return session, self._prefork_loop(session, options, options_bootstrapper)
+      prefork_fn = self._prefork_loop
     else:
-      return session, self._prefork_body(session, options, options_bootstrapper)
+      prefork_fn = self._prefork_body
+
+    target_roots, exit_code = prefork_fn(session, options, options_bootstrapper)
+    return session, target_roots, exit_code
 
   def _prefork_loop(self, session, options, options_bootstrapper):
     # TODO: See https://github.com/pantsbuild/pants/issues/6288 regarding Ctrl+C handling.
     iterations = options.for_global_scope().loop_max
     target_roots = None
+    exit_code = PANTS_SUCCEEDED_EXIT_CODE
     while iterations and not self._state.is_terminating:
       try:
-        target_roots = self._prefork_body(session, options, options_bootstrapper)
+        target_roots, exit_code = self._prefork_body(session, options, options_bootstrapper)
       except session.scheduler_session.execution_error_type as e:
         # Render retryable exceptions raised by the Scheduler.
         print(e, file=sys.stderr)
@@ -200,7 +206,7 @@ def _prefork_loop(self, session, options, options_bootstrapper):
       iterations -= 1
       while iterations and not self._state.is_terminating and not self._loop_condition.wait(timeout=1):
         continue
-    return target_roots
+    return target_roots, exit_code
 
   def _prefork_body(self, session, options, options_bootstrapper):
     global_options = options.for_global_scope()
@@ -210,22 +216,24 @@ def _prefork_body(self, session, options, options_bootstrapper):
       exclude_patterns=tuple(global_options.exclude_target_regexp) if global_options.exclude_target_regexp else tuple(),
       tags=tuple(global_options.tag) if global_options.tag else tuple()
     )
+    exit_code = PANTS_SUCCEEDED_EXIT_CODE
+
+    v1_goals, ambiguous_goals, v2_goals = options.goals_by_version
 
-    if global_options.v1:
+    if v1_goals or (ambiguous_goals and global_options.v1):
       session.warm_product_graph(target_roots)
 
-    if global_options.v2:
-      if not global_options.v1:
-        session.validate_goals(options.goals_and_possible_v2_goals)
+    if v2_goals or (ambiguous_goals and global_options.v2):
+      goals = v2_goals + (ambiguous_goals if global_options.v2 else tuple())
 
       # N.B. @console_rules run pre-fork in order to cache the products they request during execution.
-      session.run_console_rules(
+      exit_code = session.run_console_rules(
           options_bootstrapper,
-          options.goals_and_possible_v2_goals,
+          goals,
           target_roots,
         )
 
-    return target_roots
+    return target_roots, exit_code
 
   def run(self):
     """Main service entrypoint."""
diff --git a/src/python/pants/rules/core/exceptions.py b/src/python/pants/rules/core/exceptions.py
deleted file mode 100644
index 16c533fe295..00000000000
--- a/src/python/pants/rules/core/exceptions.py
+++ /dev/null
@@ -1,34 +0,0 @@
-# coding=utf-8
-# Copyright 2014 Pants project contributors (see CONTRIBUTORS.md).
-# Licensed under the Apache License, Version 2.0 (see LICENSE).
-
-from __future__ import absolute_import, division, print_function, unicode_literals
-
-from pants.base.exiter import PANTS_FAILED_EXIT_CODE, PANTS_SUCCEEDED_EXIT_CODE
-
-
-class GracefulTerminationException(Exception):
-  """Indicates that a console_rule should eagerly terminate the run.
-
-  No error trace will be printed if this is raised; the runner will simply exit with the passed
-  exit code.
-
-  Nothing except a console_rule should ever raise this.
-  """
-
-  def __init__(self, message='', exit_code=PANTS_FAILED_EXIT_CODE):
-    """
-    :param int exit_code: an optional exit code (defaults to PANTS_FAILED_EXIT_CODE)
-    """
-    super(GracefulTerminationException, self).__init__(message)
-
-    if exit_code == PANTS_SUCCEEDED_EXIT_CODE:
-      raise ValueError(
-        "Cannot create GracefulTerminationException with a successful exit code of {}"
-        .format(PANTS_SUCCEEDED_EXIT_CODE))
-
-    self._exit_code = exit_code
-
-  @property
-  def exit_code(self):
-    return self._exit_code
diff --git a/src/python/pants/rules/core/filedeps.py b/src/python/pants/rules/core/filedeps.py
index 4452a2cd9fc..6516a4a9269 100644
--- a/src/python/pants/rules/core/filedeps.py
+++ b/src/python/pants/rules/core/filedeps.py
@@ -7,28 +7,38 @@
 from pex.orderedset import OrderedSet
 
 from pants.engine.console import Console
+from pants.engine.goal import Goal, LineOriented
 from pants.engine.legacy.graph import TransitiveHydratedTargets
 from pants.engine.rules import console_rule
 
 
-@console_rule('filedeps', [Console, TransitiveHydratedTargets])
-def file_deps(console, transitive_hydrated_targets):
+class Filedeps(LineOriented, Goal):
   """List all source and BUILD files a target transitively depends on.
 
-  Files are listed with relative paths and any BUILD files implied in the transitive closure of
-  targets are also included.
+  Files may be listed with absolute or relative paths and any BUILD files implied in the transitive
+  closure of targets are also included.
   """
 
+  # TODO: Until this implements more of the options of `filedeps`, it can't claim the name!
+  name = 'fast-filedeps'
+
+
+@console_rule(Filedeps, [Console, Filedeps.Options, TransitiveHydratedTargets])
+def file_deps(console, filedeps_options, transitive_hydrated_targets):
+
   uniq_set = OrderedSet()
 
   for hydrated_target in transitive_hydrated_targets.closure:
     if hydrated_target.address.rel_path:
       uniq_set.add(hydrated_target.address.rel_path)
     if hasattr(hydrated_target.adaptor, "sources"):
-      uniq_set.update(f.path for f in hydrated_target.adaptor.sources.snapshot.files)
+      uniq_set.update(hydrated_target.adaptor.sources.snapshot.files)
+
+  with Filedeps.line_oriented(filedeps_options, console) as (print_stdout, print_stderr):
+    for f_path in uniq_set:
+      print_stdout(f_path)
 
-  for f_path in uniq_set:
-    console.print_stdout(f_path)
+  return Filedeps(exit_code=0)
 
 
 def rules():
diff --git a/src/python/pants/rules/core/fastlist.py b/src/python/pants/rules/core/list_targets.py
similarity index 70%
rename from src/python/pants/rules/core/fastlist.py
rename to src/python/pants/rules/core/list_targets.py
index b5129db2761..7ae51c99b74 100644
--- a/src/python/pants/rules/core/fastlist.py
+++ b/src/python/pants/rules/core/list_targets.py
@@ -1,5 +1,5 @@
 # coding=utf-8
-# Copyright 2018 Pants project contributors (see CONTRIBUTORS.md).
+# Copyright 2019 Pants project contributors (see CONTRIBUTORS.md).
 # Licensed under the Apache License, Version 2.0 (see LICENSE).
 
 from __future__ import absolute_import, division, print_function, unicode_literals
@@ -7,25 +7,22 @@
 from pants.base.specs import Specs
 from pants.engine.addressable import BuildFileAddresses
 from pants.engine.console import Console
+from pants.engine.goal import Goal, LineOriented
 from pants.engine.legacy.graph import HydratedTargets
-from pants.engine.rules import console_rule, optionable_rule
+from pants.engine.rules import console_rule
 from pants.engine.selectors import Get
-from pants.subsystem.subsystem import Subsystem
 
 
-class ListOptions(Subsystem):
-  """Lists all targets matching the target specs.
+class List(LineOriented, Goal):
+  """Lists all targets matching the target specs."""
 
-  If no targets are specified, lists all targets in the workspace.
-  """
+  name = 'list'
 
-  # NB: This option scope is temporary: a followup to #6880 will replace the v1 list goal and rename
-  # this scope.
-  options_scope = 'fastlist'
+  deprecated_cache_setup_removal_version = '1.18.0.dev2'
 
   @classmethod
   def register_options(cls, register):
-    super(ListOptions, cls).register_options(register)
+    super(List, cls).register_options(register)
     register('--provides', type=bool,
              help='List only targets that provide an artifact, displaying the columns specified by '
                   '--provides-columns.')
@@ -36,13 +33,11 @@ def register_options(cls, register):
              help='Print only targets that are documented with a description.')
 
 
-@console_rule('list', [Console, ListOptions, Specs])
-def fast_list(console, options, specs):
-  """A fast variant of `./pants list` with a reduced feature set."""
-
-  provides = options.get_options().provides
-  provides_columns = options.get_options().provides_columns
-  documented = options.get_options().documented
+@console_rule(List, [Console, List.Options, Specs])
+def list_targets(console, list_options, specs):
+  provides = list_options.values.provides
+  provides_columns = list_options.values.provides_columns
+  documented = list_options.values.documented
   if provides or documented:
     # To get provides clauses or documentation, we need hydrated targets.
     collection = yield Get(HydratedTargets, Specs, specs)
@@ -78,17 +73,19 @@ def print_documented(target):
     collection = yield Get(BuildFileAddresses, Specs, specs)
     print_fn = lambda address: address.spec
 
-  if not collection.dependencies:
-    console.print_stderr('WARNING: No targets were matched in goal `{}`.'.format('list'))
+  with List.line_oriented(list_options, console) as (print_stdout, print_stderr):
+    if not collection.dependencies:
+      print_stderr('WARNING: No targets were matched in goal `{}`.'.format('list'))
+
+    for item in collection:
+      result = print_fn(item)
+      if result:
+        print_stdout(result)
 
-  for item in collection:
-    result = print_fn(item)
-    if result:
-      console.print_stdout(result)
+  yield List(exit_code=0)
 
 
 def rules():
   return [
-      optionable_rule(ListOptions),
-      fast_list
+      list_targets,
     ]
diff --git a/src/python/pants/rules/core/register.py b/src/python/pants/rules/core/register.py
index 89368c0a4f9..37454785ecb 100644
--- a/src/python/pants/rules/core/register.py
+++ b/src/python/pants/rules/core/register.py
@@ -4,12 +4,8 @@
 
 from __future__ import absolute_import, division, print_function, unicode_literals
 
-from pants.rules.core import fastlist, filedeps
-from pants.rules.core.test import coordinator_of_tests, fast_test
+from pants.rules.core import filedeps, list_targets, test
 
 
 def rules():
-  return fastlist.rules() + filedeps.rules() + [
-    fast_test,
-    coordinator_of_tests,
-  ]
+  return list_targets.rules() + filedeps.rules() + test.rules()
diff --git a/src/python/pants/rules/core/test.py b/src/python/pants/rules/core/test.py
index b25d1f34eb4..d698603bdd9 100644
--- a/src/python/pants/rules/core/test.py
+++ b/src/python/pants/rules/core/test.py
@@ -5,18 +5,24 @@
 from __future__ import absolute_import, division, print_function, unicode_literals
 
 from pants.backend.python.rules.python_test_runner import PyTestResult
-from pants.base.exiter import PANTS_FAILED_EXIT_CODE
+from pants.base.exiter import PANTS_FAILED_EXIT_CODE, PANTS_SUCCEEDED_EXIT_CODE
 from pants.build_graph.address import Address
 from pants.engine.addressable import BuildFileAddresses
 from pants.engine.console import Console
+from pants.engine.goal import Goal
 from pants.engine.legacy.graph import HydratedTarget
 from pants.engine.rules import console_rule, rule
 from pants.engine.selectors import Get
 from pants.rules.core.core_test_model import Status, TestResult
-from pants.rules.core.exceptions import GracefulTerminationException
 
 
-@console_rule('test', [Console, BuildFileAddresses])
+class Test(Goal):
+  """Runs tests."""
+
+  name = 'test'
+
+
+@console_rule(Test, [Console, BuildFileAddresses])
 def fast_test(console, addresses):
   test_results = yield [Get(TestResult, Address, address.to_address()) for address in addresses]
   wrote_any_stdout = False
@@ -37,7 +43,12 @@ def fast_test(console, addresses):
     console.print_stdout('{0:80}.....{1:>10}'.format(address.reference(), test_result.status))
 
   if did_any_fail:
-    raise GracefulTerminationException("Tests failed", exit_code=PANTS_FAILED_EXIT_CODE)
+    console.print_stderr('Tests failed')
+    exit_code = PANTS_FAILED_EXIT_CODE
+  else:
+    exit_code = PANTS_SUCCEEDED_EXIT_CODE
+
+  yield Test(exit_code)
 
 
 @rule(TestResult, [HydratedTarget])
@@ -50,3 +61,10 @@ def coordinator_of_tests(target):
     yield TestResult(status=result.status, stdout=result.stdout)
   else:
     raise Exception("Didn't know how to run tests for type {}".format(target.adaptor.type_alias))
+
+
+def rules():
+  return [
+      coordinator_of_tests,
+      fast_test,
+    ]
diff --git a/src/python/pants/task/goal_options_mixin.py b/src/python/pants/task/goal_options_mixin.py
index c91ad7950ab..53bf22829fe 100644
--- a/src/python/pants/task/goal_options_mixin.py
+++ b/src/python/pants/task/goal_options_mixin.py
@@ -19,7 +19,7 @@ class GoalOptionsRegistrar(Optionable):
   at once and/or setting them per-task. E.g., turning all linters on or off, or
   turning individual linters on or off selectively.
   """
-  options_scope_category = ScopeInfo.GOAL
+  options_scope_category = ScopeInfo.GOAL_V1
 
   @classmethod
   def registrar_for_scope(cls, goal):
diff --git a/tests/python/pants_test/BUILD b/tests/python/pants_test/BUILD
index af7bd0b8845..661e867d130 100644
--- a/tests/python/pants_test/BUILD
+++ b/tests/python/pants_test/BUILD
@@ -79,6 +79,17 @@ python_library(
   ]
 )
 
+python_library(
+  name = 'console_rule_test_base',
+  sources = ['console_rule_test_base.py'],
+  dependencies = [
+    ':test_base',
+    'src/python/pants/bin',
+    'src/python/pants/init',
+  ]
+)
+
+
 python_library(
   name = 'task_test_base',
   sources = ['task_test_base.py'],
diff --git a/tests/python/pants_test/backend/graph_info/tasks/BUILD b/tests/python/pants_test/backend/graph_info/tasks/BUILD
index a753cd2e339..607cb8b59c0 100644
--- a/tests/python/pants_test/backend/graph_info/tasks/BUILD
+++ b/tests/python/pants_test/backend/graph_info/tasks/BUILD
@@ -77,7 +77,7 @@ python_tests(
     'src/python/pants/build_graph',
     'src/python/pants/backend/python/targets:python',
     'tests/python/pants_test/subsystem:subsystem_utils',
-    'tests/python/pants_test:task_test_base',
+    'tests/python/pants_test:console_rule_test_base',
   ],
 )
 
diff --git a/tests/python/pants_test/backend/graph_info/tasks/test_list_targets.py b/tests/python/pants_test/backend/graph_info/tasks/test_list_targets.py
index 2b3163953ce..7025729d0b8 100644
--- a/tests/python/pants_test/backend/graph_info/tasks/test_list_targets.py
+++ b/tests/python/pants_test/backend/graph_info/tasks/test_list_targets.py
@@ -8,7 +8,6 @@
 from builtins import object
 from textwrap import dedent
 
-from pants.backend.graph_info.tasks.list_targets import ListTargets
 from pants.backend.jvm.artifact import Artifact
 from pants.backend.jvm.repository import Repository
 from pants.backend.jvm.scala_artifact import ScalaArtifact
@@ -16,25 +15,12 @@
 from pants.backend.python.targets.python_library import PythonLibrary
 from pants.build_graph.build_file_aliases import BuildFileAliases
 from pants.build_graph.target import Target
-from pants_test.task_test_base import ConsoleTaskTestBase
+from pants.rules.core import list_targets
+from pants_test.console_rule_test_base import ConsoleRuleTestBase
 
 
-class BaseListTargetsTest(ConsoleTaskTestBase):
-
-  @classmethod
-  def task_type(cls):
-    return ListTargets
-
-
-class ListTargetsTestEmpty(BaseListTargetsTest):
-
-  def test_list_all_empty(self):
-    # NB: Also renders a warning to stderr, which is challenging to detect here but confirmed in:
-    #   tests/python/pants_test/engine/legacy/test_list_integration.py
-    self.assertEqual('', self.execute_task())
-
-
-class ListTargetsTest(BaseListTargetsTest):
+class ListTargetsTest(ConsoleRuleTestBase):
+  goal_cls = list_targets.List
 
   @classmethod
   def alias_groups(cls):
@@ -54,6 +40,10 @@ def alias_groups(cls):
       }
     )
 
+  @classmethod
+  def rules(cls):
+    return super(ListTargetsTest, cls).rules() + list_targets.rules()
+
   def setUp(self):
     super(ListTargetsTest, self).setUp()
 
@@ -96,17 +86,22 @@ def create_library(path, *libs):
         )
         '''))
 
+  def test_list_all_empty(self):
+    # NB: Also renders a warning to stderr, which is challenging to detect here but confirmed in:
+    #   tests/python/pants_test/engine/legacy/test_list_integration.py
+    self.assert_console_output(args=[])
+
   def test_list_path(self):
-    self.assert_console_output('a/b:b', targets=[self.target('a/b')])
+    self.assert_console_output('a/b:b', args=['a/b'])
 
   def test_list_siblings(self):
-    self.assert_console_output('a/b:b', targets=self.targets('a/b:'))
+    self.assert_console_output('a/b:b', args=['a/b:'])
     self.assert_console_output('a/b/c:c', 'a/b/c:c2', 'a/b/c:c3',
-                               targets=self.targets('a/b/c/:'))
+                               args=['a/b/c/:'])
 
   def test_list_descendants(self):
     self.assert_console_output('a/b/c:c', 'a/b/c:c2', 'a/b/c:c3',
-                               targets=self.targets('a/b/c/::'))
+                               args=['a/b/c/::'])
 
     self.assert_console_output(
         'a/b:b',
@@ -115,7 +110,7 @@ def test_list_descendants(self):
         'a/b/c:c3',
         'a/b/d:d',
         'a/b/e:e1',
-        targets=self.targets('a/b::'))
+        args=['a/b::'])
 
   def test_list_all(self):
     self.assert_entries('\n',
@@ -127,7 +122,7 @@ def test_list_all(self):
         'a/b/d:d',
         'a/b/e:e1',
         'f:alias',
-        targets=self.targets('::'))
+        args=['::'])
 
     self.assert_entries(', ',
         'a:a',
@@ -138,8 +133,7 @@ def test_list_all(self):
         'a/b/d:d',
         'a/b/e:e1',
         'f:alias',
-        options={'sep': ', '},
-        targets=self.targets('::'))
+        args=['--sep=, ', '::',])
 
     self.assert_console_output(
         'a:a',
@@ -150,48 +144,41 @@ def test_list_all(self):
         'a/b/d:d',
         'a/b/e:e1',
         'f:alias',
-        targets=self.targets('::'))
+        args=['::'])
 
   def test_list_provides(self):
     self.assert_console_output(
         'a/b:b com.example#b',
         'a/b/c:c2 com.example#c2',
-        options={'provides': True},
-        targets=self.targets('::'))
+        args=['--provides', '::'])
 
   def test_list_provides_customcols(self):
     self.assert_console_output(
         '/tmp a/b:b http://maven.example.com public com.example#b',
         '/tmp a/b/c:c2 http://maven.example.com public com.example#c2',
-        options={'provides': True,
-                 'provides_columns': 'push_db_basedir,address,repo_url,repo_name,artifact_id'},
-        targets=self.targets('::')
+        args=[
+          '--provides',
+          '--provides-columns=push_db_basedir,address,repo_url,repo_name,artifact_id',
+          '::',
+        ],
     )
 
   def test_list_dedups(self):
-    targets = []
-    targets.extend(self.targets('a/b/d/::'))
-    targets.extend(self.target('f:alias').dependencies)
-    self.assertEqual(3, len(targets), "Expected a duplicate of a/b/d:d")
     self.assert_console_output(
       'a/b/c:c3',
       'a/b/d:d',
-      targets=targets
+      args=['a/b/d/::', 'a/b/c:c3', 'a/b/d:d']
     )
 
   def test_list_documented(self):
     self.assert_console_output(
       # Confirm empty listing
-      targets=[self.target('a/b')],
-      options={'documented': True}
+      args=['--documented', 'a/b'],
     )
 
-    self.assert_console_output(
-      dedent("""
-      f:alias
-        Exercises alias resolution.
-        Further description.
-      """).strip(),
-      options={'documented': True},
-      targets=self.targets('::')
+    self.assert_console_output_ordered(
+      'f:alias',
+      '  Exercises alias resolution.',
+      '  Further description.',
+      args=['--documented', '::'],
     )
diff --git a/tests/python/pants_test/base/test_exception_sink_integration.py b/tests/python/pants_test/base/test_exception_sink_integration.py
index 655f109dec9..996f81a996d 100644
--- a/tests/python/pants_test/base/test_exception_sink_integration.py
+++ b/tests/python/pants_test/base/test_exception_sink_integration.py
@@ -25,10 +25,10 @@ def _assert_unhandled_exception_log_matches(self, pid, file_contents):
 process title: ([^\n]+)
 sys\\.argv: ([^\n]+)
 pid: {pid}
-Exception caught: \\(pants\\.build_graph\\.address_lookup_error\\.AddressLookupError\\)
+Exception caught: \\([^)]*\\)
 (.|\n)*
 
-Exception message: Build graph construction failed: ExecutionError 1 Exception encountered:
+Exception message:.* 1 Exception encountered:
   ResolveError: "this-target-does-not-exist" was not found in namespace ""\\. Did you mean one of:
 """.format(pid=pid))
     # Ensure we write all output such as stderr and reporting files before closing any streams.
@@ -57,8 +57,8 @@ def test_logs_unhandled_exception(self):
       self.assert_failure(pants_run)
       assertRegex(self, pants_run.stderr_data, """\
 timestamp: ([^\n]+)
-Exception caught: \\(pants\\.build_graph\\.address_lookup_error\\.AddressLookupError\\) \\(backtrace omitted\\)
-Exception message: Build graph construction failed: ExecutionError 1 Exception encountered:
+Exception caught: \\(pants\\.engine\\.scheduler\\.ExecutionError\\) \\(backtrace omitted\\)
+Exception message: 1 Exception encountered:
   ResolveError: "this-target-does-not-exist" was not found in namespace ""\\. Did you mean one of:
 """)
       pid_specific_log_file, shared_log_file = self._get_log_file_paths(tmpdir, pants_run)
diff --git a/tests/python/pants_test/console_rule_test_base.py b/tests/python/pants_test/console_rule_test_base.py
new file mode 100644
index 00000000000..7334c28b304
--- /dev/null
+++ b/tests/python/pants_test/console_rule_test_base.py
@@ -0,0 +1,134 @@
+# coding=utf-8
+# Copyright 2018 Pants project contributors (see CONTRIBUTORS.md).
+# Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+from __future__ import absolute_import, division, print_function, unicode_literals
+
+from io import StringIO
+
+from pants.engine.console import Console
+from pants.engine.goal import Goal
+from pants.engine.selectors import Params
+from pants.init.options_initializer import BuildConfigInitializer
+from pants.init.target_roots_calculator import TargetRootsCalculator
+from pants.option.options_bootstrapper import OptionsBootstrapper
+from pants.util.meta import classproperty
+from pants_test.test_base import TestBase
+
+
+class ConsoleRuleTestBase(TestBase):
+  """A baseclass useful for testing a Goal defined as a @console_rule.
+
+  :API: public
+  """
+
+  _implicit_args = tuple(['--pants-config-files=[]'])
+
+  @classproperty
+  def goal_cls(cls):
+    """Subclasses must return the Goal type to test.
+
+    :API: public
+    """
+    raise NotImplementedError()
+
+  def setUp(self):
+    super(ConsoleRuleTestBase, self).setUp()
+
+    if not issubclass(self.goal_cls, Goal):
+      raise AssertionError('goal_cls() must return a Goal subclass, got {}'.format(self.goal_cls))
+
+  def execute_rule(self, args=tuple(), env=tuple(), exit_code=0):
+    """Executes the @console_rule for this test class.
+
+    :API: public
+
+    Returns the text output of the task.
+    """
+    # Create an OptionsBootstrapper for these args/env, and a captured Console instance.
+    args = self._implicit_args + (self.goal_cls.name,) + tuple(args)
+    env = dict(env)
+    options_bootstrapper = OptionsBootstrapper.create(args=args, env=env)
+    BuildConfigInitializer.get(options_bootstrapper)
+    full_options = options_bootstrapper.get_full_options(list(self.goal_cls.Options.known_scope_infos()))
+    stdout, stderr = StringIO(), StringIO()
+    console = Console(stdout=stdout, stderr=stderr)
+
+    # Run for the target specs parsed from the args.
+    specs = TargetRootsCalculator.parse_specs(full_options.target_specs, self.build_root)
+    params = Params(specs, console, options_bootstrapper)
+    actual_exit_code = self.scheduler.run_console_rule(self.goal_cls, params)
+
+    # Flush and capture console output.
+    console.flush()
+    stdout = stdout.getvalue()
+    stderr = stderr.getvalue()
+
+    self.assertEqual(
+        exit_code,
+        actual_exit_code,
+        "Exited with {} (expected {}):\nstdout:\n{}\nstderr:\n{}".format(actual_exit_code, exit_code, stdout, stderr)
+      )
+
+    return stdout
+
+  def assert_entries(self, sep, *output, **kwargs):
+    """Verifies the expected output text is flushed by the console task under test.
+
+    NB: order of entries is not tested, just presence.
+
+    :API: public
+
+    sep:      the expected output separator.
+    *output:  the output entries expected between the separators
+    **kwargs: additional kwargs passed to execute_rule.
+    """
+    # We expect each output line to be suffixed with the separator, so for , and [1,2,3] we expect:
+    # '1,2,3,' - splitting this by the separator we should get ['1', '2', '3', ''] - always an extra
+    # empty string if the separator is properly always a suffix and not applied just between
+    # entries.
+    self.assertEqual(sorted(list(output) + ['']), sorted((self.execute_rule(**kwargs)).split(sep)))
+
+  def assert_console_output(self, *output, **kwargs):
+    """Verifies the expected output entries are emitted by the console task under test.
+
+    NB: order of entries is not tested, just presence.
+
+    :API: public
+
+    *output:  the expected output entries
+    **kwargs: additional kwargs passed to execute_rule.
+    """
+    self.assertEqual(sorted(output), sorted(self.execute_rule(**kwargs).splitlines()))
+
+  def assert_console_output_contains(self, output, **kwargs):
+    """Verifies the expected output string is emitted by the console task under test.
+
+    :API: public
+
+    output:  the expected output entry(ies)
+    **kwargs: additional kwargs passed to execute_rule.
+    """
+    self.assertIn(output, self.execute_rule(**kwargs))
+
+  def assert_console_output_ordered(self, *output, **kwargs):
+    """Verifies the expected output entries are emitted by the console task under test.
+
+    NB: order of entries is tested.
+
+    :API: public
+
+    *output:  the expected output entries in expected order
+    **kwargs: additional kwargs passed to execute_rule.
+    """
+    self.assertEqual(list(output), self.execute_rule(**kwargs).splitlines())
+
+  def assert_console_raises(self, exception, **kwargs):
+    """Verifies the expected exception is raised by the console task under test.
+
+    :API: public
+
+    **kwargs: additional kwargs are passed to execute_rule.
+    """
+    with self.assertRaises(exception):
+      self.execute_rule(**kwargs)
diff --git a/tests/python/pants_test/engine/legacy/BUILD b/tests/python/pants_test/engine/legacy/BUILD
index 308bd57441a..b39566867dc 100644
--- a/tests/python/pants_test/engine/legacy/BUILD
+++ b/tests/python/pants_test/engine/legacy/BUILD
@@ -180,16 +180,6 @@ python_tests(
   ]
 )
 
-python_tests(
-  name = 'pants_engine_integration',
-  sources = ['test_pants_engine_integration.py'],
-  dependencies = [
-    'tests/python/pants_test:int-test',
-    'tests/python/pants_test/testutils:py2_compat',
-  ],
-  tags = {'integration'}
-)
-
 python_tests(
   name = 'parser',
   sources = ['test_parser.py'],
diff --git a/tests/python/pants_test/engine/legacy/test_build_ignore_integration.py b/tests/python/pants_test/engine/legacy/test_build_ignore_integration.py
index 85fe8bdfc88..b3eb7cba798 100644
--- a/tests/python/pants_test/engine/legacy/test_build_ignore_integration.py
+++ b/tests/python/pants_test/engine/legacy/test_build_ignore_integration.py
@@ -71,7 +71,7 @@ def output_to_list(output_filename):
 
   def test_build_ignore_dependency(self):
     run_result = self.run_pants(['-q',
-                                 'list',
+                                 'dependencies',
                                  'testprojects/tests/python/pants::'],
                                 config={
                                   'DEFAULT': {
@@ -87,7 +87,7 @@ def test_build_ignore_dependency(self):
 
   def test_build_ignore_dependency_success(self):
     run_result = self.run_pants(['-q',
-                                 'list',
+                                 'dependencies',
                                  'testprojects/tests/python/pants::'],
                                 config={
                                   'DEFAULT': {
diff --git a/tests/python/pants_test/engine/legacy/test_console_rule_integration.py b/tests/python/pants_test/engine/legacy/test_console_rule_integration.py
index 7c274b5fec8..164892eb21a 100644
--- a/tests/python/pants_test/engine/legacy/test_console_rule_integration.py
+++ b/tests/python/pants_test/engine/legacy/test_console_rule_integration.py
@@ -51,14 +51,14 @@ def test_v2_goal_validation(self):
     result = self.do_command(
       '--no-v1',
       '--v2',
-      'lint',
+      'blah',
       '::',
       success=False
     )
 
     self.assertIn(
-      'could not satisfy the following goals with @console_rules: lint',
-      result.stderr_data
+      'Unknown goals: blah',
+      result.stdout_data
     )
 
   @ensure_daemon
diff --git a/tests/python/pants_test/engine/legacy/test_graph_integration.py b/tests/python/pants_test/engine/legacy/test_graph_integration.py
index b09a39a7922..bbcfdba6308 100644
--- a/tests/python/pants_test/engine/legacy/test_graph_integration.py
+++ b/tests/python/pants_test/engine/legacy/test_graph_integration.py
@@ -9,7 +9,6 @@
 
 from future.utils import PY2
 
-from pants.build_graph.address_lookup_error import AddressLookupError
 from pants.option.scope import GLOBAL_SCOPE_CONFIG_SECTION
 from pants_test.pants_run_integration_test import PantsRunIntegrationTest
 
@@ -53,7 +52,7 @@ def _list_target_check_warnings_sources(self, target_name):
     target_full = '{}:{}'.format(self._SOURCES_TARGET_BASE, target_name)
     glob_str, expected_globs = self._SOURCES_ERR_MSGS[target_name]
 
-    pants_run = self.run_pants(['list', target_full], config={
+    pants_run = self.run_pants(['filedeps', target_full], config={
       GLOBAL_SCOPE_CONFIG_SECTION: {
         'glob_expansion_failure': 'warn',
       },
@@ -93,15 +92,13 @@ def _list_target_check_warnings_sources(self, target_name):
   def _list_target_check_error(self, target_name):
     expected_excerpts = self._ERR_TARGETS[target_name]
 
-    pants_run = self.run_pants(['list', target_name], config={
+    pants_run = self.run_pants(['filedeps', target_name], config={
       GLOBAL_SCOPE_CONFIG_SECTION: {
         'glob_expansion_failure': 'error',
       },
     })
     self.assert_failure(pants_run)
 
-    self.assertIn(AddressLookupError.__name__, pants_run.stderr_data)
-
     for excerpt in expected_excerpts:
       self.assertIn(excerpt, pants_run.stderr_data)
 
@@ -111,7 +108,7 @@ def test_missing_sources_warnings(self):
 
   def test_existing_sources(self):
     target_full = '{}:text'.format(self._SOURCES_TARGET_BASE)
-    pants_run = self.run_pants(['list', target_full], config={
+    pants_run = self.run_pants(['filedeps', target_full], config={
       GLOBAL_SCOPE_CONFIG_SECTION: {
         'glob_expansion_failure': 'warn',
       },
@@ -121,7 +118,7 @@ def test_existing_sources(self):
 
   def test_missing_bundles_warnings(self):
     target_full = '{}:missing-bundle-fileset'.format(self._BUNDLE_TARGET_BASE)
-    pants_run = self.run_pants(['list', target_full], config={
+    pants_run = self.run_pants(['filedeps', target_full], config={
       GLOBAL_SCOPE_CONFIG_SECTION: {
         'glob_expansion_failure': 'warn',
       },
@@ -140,7 +137,7 @@ def test_missing_bundles_warnings(self):
 
   def test_existing_bundles(self):
     target_full = '{}:mapper'.format(self._BUNDLE_TARGET_BASE)
-    pants_run = self.run_pants(['list', target_full], config={
+    pants_run = self.run_pants(['filedeps', target_full], config={
       GLOBAL_SCOPE_CONFIG_SECTION: {
         'glob_expansion_failure': 'warn',
       },
diff --git a/tests/python/pants_test/engine/legacy/test_pants_engine_integration.py b/tests/python/pants_test/engine/legacy/test_pants_engine_integration.py
deleted file mode 100644
index 1d98f585341..00000000000
--- a/tests/python/pants_test/engine/legacy/test_pants_engine_integration.py
+++ /dev/null
@@ -1,24 +0,0 @@
-# coding=utf-8
-# Copyright 2016 Pants project contributors (see CONTRIBUTORS.md).
-# Licensed under the Apache License, Version 2.0 (see LICENSE).
-
-from __future__ import absolute_import, division, print_function, unicode_literals
-
-from pants_test.pants_run_integration_test import PantsRunIntegrationTest
-from pants_test.testutils.py2_compat import assertRegex
-
-
-class PantsEngineIntegrationTest(PantsRunIntegrationTest):
-  def test_engine_list(self):
-    pants_run = self.run_pants(['-ldebug', 'list', '3rdparty::'])
-    self.assert_success(pants_run)
-    assertRegex(self, pants_run.stderr_data, 'build_graph is: .*LegacyBuildGraph')
-    assertRegex(self, pants_run.stderr_data,
-                             'computed \d+ nodes in')
-
-  def test_engine_binary(self):
-    self.assert_success(
-      self.run_pants(
-        ['binary', 'examples/src/python/example/hello/main:']
-      )
-    )
diff --git a/tests/python/pants_test/engine/test_rules.py b/tests/python/pants_test/engine/test_rules.py
index 7a182dbd114..8da213c74c6 100644
--- a/tests/python/pants_test/engine/test_rules.py
+++ b/tests/python/pants_test/engine/test_rules.py
@@ -13,8 +13,9 @@
 from pants.engine.build_files import create_graph_rules
 from pants.engine.console import Console
 from pants.engine.fs import create_fs_rules
+from pants.engine.goal import Goal
 from pants.engine.mapper import AddressMapper
-from pants.engine.rules import RootRule, RuleIndex, _GoalProduct, _RuleVisitor, console_rule, rule
+from pants.engine.rules import RootRule, RuleIndex, _RuleVisitor, console_rule, rule
 from pants.engine.selectors import Get
 from pants_test.engine.examples.parsers import JsonParser
 from pants_test.engine.util import (TARGET_TABLE, assert_equal_with_printing, create_scheduler,
@@ -62,10 +63,17 @@ def __repr__(self):
 _this_is_not_a_type = 3
 
 
-@console_rule('example', [Console])
+class Example(Goal):
+  """An example."""
+
+  name = 'example'
+
+
+@console_rule(Example, [Console])
 def a_console_rule_generator(console):
   a = yield Get(A, str('a str!'))
   console.print_stdout(str(a))
+  yield Example(exit_code=0)
 
 
 class RuleTest(unittest.TestCase):
@@ -73,7 +81,7 @@ def test_run_rule_console_rule_generator(self):
     res = run_rule(a_console_rule_generator, Console(), {
         (A, str): lambda _: A(),
       })
-    self.assertEquals(res, _GoalProduct.for_name('example')())
+    self.assertEquals(res, Example(0))
 
 
 class RuleIndexTest(TestBase):
diff --git a/tests/python/pants_test/pantsd/pantsd_integration_test_base.py b/tests/python/pants_test/pantsd/pantsd_integration_test_base.py
index b278a819456..94b9607a65c 100644
--- a/tests/python/pants_test/pantsd/pantsd_integration_test_base.py
+++ b/tests/python/pants_test/pantsd/pantsd_integration_test_base.py
@@ -115,11 +115,6 @@ def pantsd_test_context(self, log_level='info', extra_config=None):
         self.assert_runner(workdir, pantsd_config, ['kill-pantsd'], expected_runs=1)
         try:
           yield workdir, pantsd_config, checker
-        finally:
-          banner('BEGIN pantsd.log')
-          for line in read_pantsd_log(workdir):
-            print(line)
-          banner('END pantsd.log')
           self.assert_runner(
             workdir,
             pantsd_config,
@@ -127,6 +122,11 @@ def pantsd_test_context(self, log_level='info', extra_config=None):
             expected_runs=1,
           )
           checker.assert_stopped()
+        finally:
+          banner('BEGIN pantsd.log')
+          for line in read_pantsd_log(workdir):
+            print(line)
+          banner('END pantsd.log')
 
   @contextmanager
   def pantsd_successful_run_context(self, *args, **kwargs):
diff --git a/tests/python/pants_test/rules/BUILD b/tests/python/pants_test/rules/BUILD
index ccadf9196bc..4125fa3c6a2 100644
--- a/tests/python/pants_test/rules/BUILD
+++ b/tests/python/pants_test/rules/BUILD
@@ -1,8 +1,13 @@
+# Copyright 2019 Pants project contributors (see CONTRIBUTORS.md).
+# Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+INTEGRATION_SOURCES = globs('*_integration.py')
+
 python_tests(
-  name='test',
-  sources=['test_test.py'],
+  sources=globs('*.py', exclude=[INTEGRATION_SOURCES]),
   dependencies=[
     '3rdparty/python:future',
+    '3rdparty/python:mock',
     'src/python/pants/backend/python/rules',
     'src/python/pants/util:objects',
     'tests/python/pants_test:test_base',
@@ -12,8 +17,8 @@ python_tests(
 )
 
 python_tests(
-  name='test_integration',
-  sources=['test_test_integration.py'],
+  name='integration',
+  sources=INTEGRATION_SOURCES,
   dependencies=[
     'tests/python/pants_test:int-test',
   ],
diff --git a/tests/python/pants_test/rules/test_filedeps.py b/tests/python/pants_test/rules/test_filedeps.py
index c609b535cfc..17166ecb358 100644
--- a/tests/python/pants_test/rules/test_filedeps.py
+++ b/tests/python/pants_test/rules/test_filedeps.py
@@ -4,6 +4,7 @@
 
 from __future__ import absolute_import, division, print_function, unicode_literals
 
+import unittest
 from textwrap import dedent
 
 from mock import Mock
@@ -16,6 +17,7 @@
 from pants_test.test_base import TestBase
 
 
+@unittest.skip(reason='Bitrot discovered during #6880: should be ported to ConsoleRuleTestBase.')
 class FileDepsTest(TestBase):
 
   def filedeps_rule_test(self, transitive_targets, expected_console_output):
diff --git a/tests/python/pants_test/rules/test_filedeps_integration.py b/tests/python/pants_test/rules/test_filedeps_integration.py
index 6e753dbffd1..31c0fdf4eac 100644
--- a/tests/python/pants_test/rules/test_filedeps_integration.py
+++ b/tests/python/pants_test/rules/test_filedeps_integration.py
@@ -15,7 +15,7 @@ def test_filedeps_multiple_targets_with_dep(self):
     args = [
       '--no-v1',
       '--v2',
-      'filedeps',
+      'fast-filedeps',
       'examples/src/scala/org/pantsbuild/example/hello/exe:exe',
       'examples/src/scala/org/pantsbuild/example/hello/welcome:welcome'
     ]
diff --git a/tests/python/pants_test/rules/test_test.py b/tests/python/pants_test/rules/test_test.py
index d8be8f72f90..15869a46842 100644
--- a/tests/python/pants_test/rules/test_test.py
+++ b/tests/python/pants_test/rules/test_test.py
@@ -10,7 +10,6 @@
 from pants.build_graph.address import Address, BuildFileAddress
 from pants.engine.legacy.graph import HydratedTarget
 from pants.engine.legacy.structs import PythonTestsAdaptor
-from pants.rules.core.exceptions import GracefulTerminationException
 from pants.rules.core.test import Status, TestResult, coordinator_of_tests, fast_test
 from pants.util.meta import AbstractClass
 from pants_test.engine.scheduler_test_base import SchedulerTestBase
@@ -19,14 +18,15 @@
 
 
 class TestTest(TestBase, SchedulerTestBase, AbstractClass):
-  def single_target_test(self, result, expected_console_output):
+  def single_target_test(self, result, expected_console_output, success=True):
     console = MockConsole()
 
-    run_rule(fast_test, console, (self.make_build_target_address("some/target"),), {
+    res = run_rule(fast_test, console, (self.make_build_target_address("some/target"),), {
       (TestResult, Address): lambda _: result,
     })
 
     self.assertEquals(console.stdout.getvalue(), expected_console_output)
+    self.assertEquals(0 if success else 1, res.exit_code)
 
   def make_build_target_address(self, spec):
     address = Address.parse(spec)
@@ -46,15 +46,14 @@ def test_outputs_success(self):
     )
 
   def test_output_failure(self):
-    with self.assertRaises(GracefulTerminationException) as cm:
-      self.single_target_test(
-        TestResult(status=Status.FAILURE, stdout='Here is some output from a test'),
-        """Here is some output from a test
+    self.single_target_test(
+      TestResult(status=Status.FAILURE, stdout='Here is some output from a test'),
+      """Here is some output from a test
 
 some/target                                                                     .....   FAILURE
-"""
-      )
-    self.assertEqual(1, cm.exception.exit_code)
+""",
+      success=False,
+    )
 
   def test_output_no_trailing_newline(self):
     self.single_target_test(
@@ -87,12 +86,11 @@ def make_result(target):
       else:
         raise Exception("Unrecognised target")
 
-    with self.assertRaises(GracefulTerminationException) as cm:
-      run_rule(fast_test, console, (target1, target2), {
-        (TestResult, Address): make_result,
-      })
+    res = run_rule(fast_test, console, (target1, target2), {
+      (TestResult, Address): make_result,
+    })
 
-    self.assertEqual(1, cm.exception.exit_code)
+    self.assertEqual(1, res.exit_code)
     self.assertEquals(console.stdout.getvalue(), """I passed
 I failed
 
diff --git a/tests/python/pants_test/rules/test_test_integration.py b/tests/python/pants_test/rules/test_test_integration.py
index 55572f16d82..7b9f61ced11 100644
--- a/tests/python/pants_test/rules/test_test_integration.py
+++ b/tests/python/pants_test/rules/test_test_integration.py
@@ -50,6 +50,8 @@ def run_failing_pants_test(self, trailing_args):
 
     pants_run = self.run_pants(args)
     self.assert_failure(pants_run)
+    self.assertEqual('Tests failed\n', pants_run.stderr_data)
+
     return pants_run
 
   def test_passing_python_test(self):