diff --git a/build-support/bin/release.sh b/build-support/bin/release.sh
index f7e83c9fca1b..281cf878e9ed 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 c4514906daf3..c5728f94400e 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
@@ -8,7 +8,6 @@
 from pants.engine.console import Console
 from pants.engine.goal import Goal
 from pants.engine.rules import console_rule
-from pants.rules.core.exceptions import GracefulTerminationException
 
 
 class ListAndDieForTesting(Goal):
@@ -21,7 +20,7 @@ class ListAndDieForTesting(Goal):
 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():
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 4e59fd816d17..62aec840e7d9 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,6 +9,7 @@
 
 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
@@ -16,7 +17,6 @@
 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
@@ -210,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, Validate])
-def validate(console, hydrated_targets, validate_goal):
+@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 = validate_goal.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
@@ -241,7 +241,11 @@ def validate(console, hydrated_targets, validate_goal):
       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/backend/project_info/tasks/filedeps.py b/src/python/pants/backend/project_info/tasks/filedeps.py
index d9cf94659757..aa1b37aecc43 100644
--- a/src/python/pants/backend/project_info/tasks/filedeps.py
+++ b/src/python/pants/backend/project_info/tasks/filedeps.py
@@ -9,15 +9,23 @@
 from pants.backend.jvm.targets.jvm_app import JvmApp
 from pants.backend.jvm.targets.scala_library import ScalaLibrary
 from pants.base.build_environment import get_buildroot
-from pants.rules.core import filedeps as filedeps_rules
 from pants.task.console_task import ConsoleTask
 
 
 class FileDeps(ConsoleTask):
+  """List all source and BUILD files a target transitively depends on.
+
+  Files may be listed with absolute or relative paths and any BUILD files implied in the transitive
+  closure of targets are also included.
+  """
 
   @classmethod
-  def subsystem_dependencies(cls):
-    return super(FileDeps, cls).subsystem_dependencies() + (filedeps_rules.Filedeps,)
+  def register_options(cls, register):
+    super(FileDeps, cls).register_options(register)
+    register('--globs', type=bool,
+             help='Instead of outputting filenames, output globs (ignoring excludes)')
+    register('--absolute', type=bool, default=True,
+             help='If True output with absolute path, else output with path relative to the build root')
 
   def _file_path(self, path):
     return os.path.join(get_buildroot(), path) if self.get_options().absolute else path
diff --git a/src/python/pants/bin/daemon_pants_runner.py b/src/python/pants/bin/daemon_pants_runner.py
index fb88ae593451..be777abf4a5f 100644
--- a/src/python/pants/bin/daemon_pants_runner.py
+++ b/src/python/pants/bin/daemon_pants_runner.py
@@ -23,7 +23,6 @@
 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
 
@@ -92,7 +91,7 @@ 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)
+        graph_helper, target_roots, _ = scheduler_service.prefork(options, options_bootstrapper)
         deferred_exc = None
     except Exception:
       deferred_exc = sys.exc_info()
@@ -332,10 +331,6 @@ def post_fork_child(self):
         runner.run()
       except KeyboardInterrupt:
         self._exiter.exit_and_fail('Interrupted by user.\n')
-      except GracefulTerminationException as e:
-        ExceptionSink.log_exception(
-          'Encountered graceful termination exception {}; exiting'.format(e))
-        self._exiter.exit(e.exit_code)
       except Exception:
         # TODO: We override sys.excepthook above when we call ExceptionSink.set_exiter(). That
         # excepthook catches `SignalHandledNonLocalExit`s from signal handlers, which isn't
diff --git a/src/python/pants/bin/local_pants_runner.py b/src/python/pants/bin/local_pants_runner.py
index d66bf1196913..7b05205374c7 100644
--- a/src/python/pants/bin/local_pants_runner.py
+++ b/src/python/pants/bin/local_pants_runner.py
@@ -23,7 +23,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
 
 
@@ -273,14 +272,11 @@ def _maybe_run_v2(self):
       return PANTS_SUCCEEDED_EXIT_CODE
 
     try:
-      self._graph_session.run_console_rules(
+      return self._graph_session.run_console_rules(
         self._options_bootstrapper,
         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())
diff --git a/src/python/pants/engine/goal.py b/src/python/pants/engine/goal.py
index 7be504a9603a..10531af06f7f 100644
--- a/src/python/pants/engine/goal.py
+++ b/src/python/pants/engine/goal.py
@@ -10,14 +10,23 @@
 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(SubsystemClientMixin, Optionable, AbstractClass):
-  """A CLI goal whch is implemented by a `@console_rule`.
+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.
   """
 
   # Subclasser-defined. See the class pydoc.
@@ -28,48 +37,76 @@ class Goal(SubsystemClientMixin, Optionable, AbstractClass):
   # that association.
   deprecated_cache_setup_removal_version = None
 
-  @classproperty
-  def options_scope(cls):
-    if not cls.name:
-      # TODO: Would it be unnecessarily magical to have `cls.__name__.lower()` always be the name?
-      raise AssertionError('{} must have a `Goal.name` defined.'.format(cls.__name__))
-    return cls.name
-
   @classmethod
-  def subsystem_dependencies(cls):
-    # NB: `Goal` 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
-    # Subsystem deps.
-    if cls.deprecated_cache_setup_removal_version:
-      dep = CacheSetup.scoped(
-          cls,
-          removal_version=cls.deprecated_cache_setup_removal_version,
-          removal_hint='Goal `{}` uses an independent caching implementation, and ignores `{}`.'.format(
-            cls.name,
-            CacheSetup.subscope(cls.name),
-          )
-        )
-      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(Goal, self).__init__()
-    self._scope = scope
-    self._scoped_options = scoped_options
-
-  @property
-  def options(self):
-    """Returns the option values for this Goal."""
-    return self._scoped_options
+  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):
+        if not outer_cls.name:
+          # TODO: Would it be unnecessarily magical to have `outer_cls.__name__.lower()` always be
+          # the name?
+          raise AssertionError('{} must have a `Goal.name` defined.'.format(outer_cls.__name__))
+        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
+    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 and a context manager for line-oriented output."""
+  """A mixin for Goal that adds Options to support the `line_oriented` context manager."""
 
   @classmethod
   def register_options(cls, register):
@@ -79,24 +116,32 @@ def register_options(cls, register):
     register('--output-file', metavar='<path>',
              help='Write line-oriented output to this file instead.')
 
-  @contextmanager
-  def line_oriented(self, console):
-    """Takes a Console and yields functions for writing to stdout and stderr, respectively."""
 
-    output_file = self.options.output_file
-    sep = self.options.sep.encode('utf-8').decode('unicode_escape')
+@contextmanager
+def line_oriented(line_oriented_options, console):
+  """Given options and a Console, yields functions for writing to stdout and stderr, respectively.
 
-    stdout, stderr = console.stdout, console.stderr
+  The passed options instance will generally be the `Goal.Options` of a `LineOriented` `Goal`.
+  """
+  if not isinstance(line_oriented_options, _GoalOptions):
+    raise AssertionError(
+        'Expected a `Goal.Options` instance for a `LineOriented` `Goal`, got: {}'.format(
+          type(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 = 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()
+      stdout.close()
+    else:
+      stdout.flush()
+    stderr.flush()
diff --git a/src/python/pants/engine/rules.py b/src/python/pants/engine/rules.py
index 2368f018bbed..413e6f753451 100644
--- a/src/python/pants/engine/rules.py
+++ b/src/python/pants/engine/rules.py
@@ -5,17 +5,13 @@
 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
@@ -175,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.
@@ -231,15 +191,22 @@ def _get_starting_indent(source):
   return 0
 
 
-def _make_rule(output_type, input_selectors, goal_cls=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 Goal goal_cls: If this is a `@console_rule`, a Goal class to provide options, help, and a scope.
   """
 
+  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.')
@@ -283,34 +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 goal_cls:
-      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
-      dependency_rules = (optionable_rule(goal_cls),)
+    # 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_cls=goal_cls,
         dependency_rules=dependency_rules,
         cacheable=cacheable,
       )
 
-    return wrapped_func
+    return func
   return wrapper
 
 
@@ -319,11 +274,7 @@ def rule(output_type, input_selectors):
 
 
 def console_rule(goal_cls, input_selectors):
-  if not isinstance(goal_cls, type) or not issubclass(goal_cls, Goal):
-    raise TypeError('The first argument for a @console_rule must be an associated `Goal` to '
-                    'declare its help, options, and scope. Got: `{}`.'.format(goal_cls))
-  output_type = _GoalProduct.for_name(goal_cls.options_scope)
-  return _make_rule(output_type, input_selectors, goal_cls, False)
+  return _make_rule(goal_cls, input_selectors, False)
 
 
 def union(cls):
@@ -400,7 +351,6 @@ class TaskRule(datatype([
   ('input_selectors', TypedCollection(SubclassesOf(type))),
   ('input_gets', tuple),
   'func',
-  'goal_cls',
   ('dependency_rules', tuple),
   ('dependency_optionables', tuple),
   ('cacheable', bool),
@@ -419,14 +369,8 @@ def __new__(cls,
               input_gets,
               dependency_optionables=None,
               dependency_rules=None,
-              goal_cls=None,
               cacheable=True):
 
-    # Validate associated Goal (if any).
-    if goal_cls and not issubclass(goal_cls, Goal):
-      raise TypeError("Expected a Goal instance for `{}`, got: {}".format(
-        func.__name__, type(goal_cls)))
-
     # Create.
     return super(TaskRule, cls).__new__(
         cls,
@@ -434,7 +378,6 @@ def __new__(cls,
         input_selectors,
         input_gets,
         func,
-        goal_cls,
         dependency_rules or tuple(),
         dependency_optionables or tuple(),
         cacheable,
diff --git a/src/python/pants/engine/scheduler.py b/src/python/pants/engine/scheduler.py
index b5dfd0c3ba94..843bd202405f 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/init/engine_initializer.py b/src/python/pants/init/engine_initializer.py
index d168ea4bc374..3f90ad4c7c76 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,
@@ -174,6 +176,8 @@ 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.
     """
     subject = target_roots.specs
     console = Console()
@@ -182,10 +186,14 @@ def run_console_rules(self, options_bootstrapper, goals, target_roots):
       params = Params(subject, options_bootstrapper, console)
       logger.debug('requesting {} to satisfy execution of `{}` goal'.format(goal_product, goal))
       try:
-        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.
 
@@ -215,9 +223,10 @@ class GoalMappingError(Exception):
   def _make_goal_map_from_rules(rules):
     goal_map = {}
     for r in rules:
-      if getattr(r, 'goal_cls', None) is None:
+      output_type = getattr(r, 'output_type', None)
+      if not output_type or not issubclass(output_type, Goal):
         continue
-      goal = r.goal_cls.options_scope
+      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/pantsd/service/scheduler_service.py b/src/python/pants/pantsd/service/scheduler_service.py
index 59c838ebda14..ff02e2ab8220 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,6 +216,7 @@ 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
 
@@ -220,13 +227,13 @@ def _prefork_body(self, session, options, options_bootstrapper):
       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,
           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 16c533fe295e..000000000000
--- 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 1b39fac113dc..c6ec2ad4846f 100644
--- a/src/python/pants/rules/core/filedeps.py
+++ b/src/python/pants/rules/core/filedeps.py
@@ -7,31 +7,24 @@
 from pex.orderedset import OrderedSet
 
 from pants.engine.console import Console
-from pants.engine.goal import Goal
+from pants.engine.goal import Goal, LineOriented, line_oriented
 from pants.engine.legacy.graph import TransitiveHydratedTargets
 from pants.engine.rules import console_rule
 
 
-class Filedeps(Goal):
+class Filedeps(LineOriented, Goal):
   """List all source and BUILD files a target transitively depends on.
 
   Files may be listed with absolute or relative paths and any BUILD files implied in the transitive
   closure of targets are also included.
   """
 
-  name = 'filedeps'
+  # TODO: Until this implements more of the options of `filedeps`, it can't claim the name!
+  name = 'fast-filedeps'
 
-  @classmethod
-  def register_options(cls, register):
-    super(Filedeps, cls).register_options(register)
-    register('--globs', type=bool,
-             help='Instead of outputting filenames, output globs (ignoring excludes)')
-    register('--absolute', type=bool, default=True,
-             help='If True output with absolute path, else output with path relative to the build root')
 
-
-@console_rule(Filedeps, [Console, TransitiveHydratedTargets])
-def file_deps(console, transitive_hydrated_targets):
+@console_rule(Filedeps, [Console, Filedeps.Options, TransitiveHydratedTargets])
+def file_deps(console, filedeps_options, transitive_hydrated_targets):
 
   uniq_set = OrderedSet()
 
@@ -39,10 +32,13 @@ def file_deps(console, transitive_hydrated_targets):
     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(f for f in hydrated_target.adaptor.sources.snapshot.files)
+
+  with 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/list_targets.py b/src/python/pants/rules/core/list_targets.py
index 9c50df9648b9..c6801364d18b 100644
--- a/src/python/pants/rules/core/list_targets.py
+++ b/src/python/pants/rules/core/list_targets.py
@@ -7,7 +7,7 @@
 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.goal import Goal, LineOriented, line_oriented
 from pants.engine.legacy.graph import HydratedTargets
 from pants.engine.rules import console_rule
 from pants.engine.selectors import Get
@@ -33,11 +33,11 @@ def register_options(cls, register):
              help='Print only targets that are documented with a description.')
 
 
-@console_rule(List, [Console, List, Specs])
-def list_targets(console, list_goal, specs):
-  provides = list_goal.options.provides
-  provides_columns = list_goal.options.provides_columns
-  documented = list_goal.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)
@@ -73,7 +73,7 @@ def print_documented(target):
     collection = yield Get(BuildFileAddresses, Specs, specs)
     print_fn = lambda address: address.spec
 
-  with list_goal.line_oriented(console) as (print_stdout, print_stderr):
+  with 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'))
 
@@ -82,6 +82,8 @@ def print_documented(target):
       if result:
         print_stdout(result)
 
+  yield List(exit_code=0)
+
 
 def rules():
   return [
diff --git a/src/python/pants/rules/core/test.py b/src/python/pants/rules/core/test.py
index 142f164e72b5..d698603bdd9f 100644
--- a/src/python/pants/rules/core/test.py
+++ b/src/python/pants/rules/core/test.py
@@ -5,7 +5,7 @@
 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
@@ -14,7 +14,6 @@
 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
 
 
 class Test(Goal):
@@ -44,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])
diff --git a/tests/python/pants_test/console_rule_test_base.py b/tests/python/pants_test/console_rule_test_base.py
index 05f70d22bdf8..7334c28b3043 100644
--- a/tests/python/pants_test/console_rule_test_base.py
+++ b/tests/python/pants_test/console_rule_test_base.py
@@ -8,7 +8,6 @@
 
 from pants.engine.console import Console
 from pants.engine.goal import Goal
-from pants.engine.rules import _GoalProduct
 from pants.engine.selectors import Params
 from pants.init.options_initializer import BuildConfigInitializer
 from pants.init.target_roots_calculator import TargetRootsCalculator
@@ -39,7 +38,7 @@ def setUp(self):
     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()):
+  def execute_rule(self, args=tuple(), env=tuple(), exit_code=0):
     """Executes the @console_rule for this test class.
 
     :API: public
@@ -51,19 +50,27 @@ def execute_rule(self, args=tuple(), env=tuple()):
     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.known_scope_infos()))
+    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)
-    goal_product = _GoalProduct.for_name(self.goal_cls.options_scope)
-    self.scheduler.run_console_rule(goal_product, params)
+    actual_exit_code = self.scheduler.run_console_rule(self.goal_cls, params)
 
     # Flush and capture console output.
     console.flush()
-    return stdout.getvalue()
+    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.
diff --git a/tests/python/pants_test/engine/test_rules.py b/tests/python/pants_test/engine/test_rules.py
index cf0899ca5309..8da213c74c6f 100644
--- a/tests/python/pants_test/engine/test_rules.py
+++ b/tests/python/pants_test/engine/test_rules.py
@@ -15,7 +15,7 @@
 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,
@@ -73,6 +73,7 @@ class Example(Goal):
 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):
@@ -80,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/rules/BUILD b/tests/python/pants_test/rules/BUILD
index ccadf9196bca..4125fa3c6a2b 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 c609b535cfc7..17166ecb3585 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 6e753dbffd1d..31c0fdf4eacd 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 d8be8f72f90a..15869a468421 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 55572f16d828..7b9f61ced112 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):