diff --git a/src/python/pants/base/exception_sink.py b/src/python/pants/base/exception_sink.py
index 8fd1af9c9cdf..2f55fd3da2a0 100644
--- a/src/python/pants/base/exception_sink.py
+++ b/src/python/pants/base/exception_sink.py
@@ -10,12 +10,13 @@
 import threading
 import traceback
 from contextlib import contextmanager
-from typing import Callable, Iterator, Optional
+from typing import Callable, Optional
 
 import setproctitle
 
-from pants.base.exiter import Exiter
+from pants.base.exiter import ExitCode
 from pants.util.dirutil import safe_mkdir, safe_open
+from pants.util.meta import classproperty
 from pants.util.osutil import Pid
 
 logger = logging.getLogger(__name__)
@@ -112,7 +113,7 @@ class ExceptionSink:
     _log_dir = None
     # We need an exiter in order to know what to do after we log a fatal exception or handle a
     # catchable signal.
-    _exiter: Optional[Exiter] = None
+    _exiter: Callable[[ExitCode], None] = sys.exit
     # Where to log stacktraces to in a SIGUSR2 handler.
     _interactive_output_stream = None
     # Whether to print a stacktrace in any fatal error message printed to the terminal.
@@ -133,6 +134,11 @@ def __new__(cls, *args, **kwargs):
     class ExceptionSinkError(Exception):
         pass
 
+    @classmethod
+    def exit(cls, code: ExitCode):
+        """Exits using the currently configured global exiter."""
+        cls._exiter(code)
+
     @classmethod
     def reset_should_print_backtrace_to_terminal(cls, should_print_backtrace):
         """Set whether a backtrace gets printed to the terminal error stream on a fatal error.
@@ -198,47 +204,8 @@ def reset_log_location(cls, new_log_location: str) -> None:
         cls._pid_specific_error_fileobj = pid_specific_error_stream
         cls._shared_error_fileobj = shared_error_stream
 
-    class AccessGlobalExiterMixin:
-        @property
-        def _exiter(self) -> Optional[Exiter]:
-            return ExceptionSink.get_global_exiter()
-
-    @classmethod
-    def get_global_exiter(cls) -> Optional[Exiter]:
-        return cls._exiter
-
-    @classmethod
-    @contextmanager
-    def exiter_as(cls, new_exiter_fun: Callable[[Optional[Exiter]], Exiter]) -> Iterator[None]:
-        """Temporarily override the global exiter.
-
-        NB: We don't want to try/finally here, because we want exceptions to propagate
-        with the most recent exiter installed in sys.excepthook.
-        If we wrap this in a try:finally, exceptions will be caught and exiters unset.
-        """
-        previous_exiter = cls._exiter
-        new_exiter = new_exiter_fun(previous_exiter)
-        cls._reset_exiter(new_exiter)
-        yield
-        cls._reset_exiter(previous_exiter)
-
-    @classmethod
-    @contextmanager
-    def exiter_as_until_exception(
-        cls, new_exiter_fun: Callable[[Optional[Exiter]], Exiter]
-    ) -> Iterator[None]:
-        """Temporarily override the global exiter, except this will unset it when an exception
-        happens."""
-        previous_exiter = cls._exiter
-        new_exiter = new_exiter_fun(previous_exiter)
-        try:
-            cls._reset_exiter(new_exiter)
-            yield
-        finally:
-            cls._reset_exiter(previous_exiter)
-
     @classmethod
-    def _reset_exiter(cls, exiter: Optional[Exiter]) -> None:
+    def reset_exiter(cls, exiter: Callable[[ExitCode], None]) -> None:
         """Class state:
 
         - Overwrites `cls._exiter`.
@@ -250,7 +217,7 @@ def _reset_exiter(cls, exiter: Optional[Exiter]) -> None:
         # uncaught exception handler uses cls._exiter to exit.
         cls._exiter = exiter
         # NB: mutate process-global state!
-        sys.excepthook = cls._log_unhandled_exception_and_exit
+        sys.excepthook = cls.log_exception
 
     @classmethod
     def reset_interactive_output_stream(
@@ -276,10 +243,14 @@ def reset_interactive_output_stream(
             cls._interactive_output_stream = interactive_output_stream
         except ValueError:
             # Warn about "ValueError: IO on closed file" when the stream is closed.
-            cls.log_exception(
+            cls._log_exception(
                 "Cannot reset interactive_output_stream -- stream (probably stderr) is closed"
             )
 
+    @classproperty
+    def should_print_exception_stacktrace(cls):
+        return cls._should_print_backtrace_to_terminal
+
     @classmethod
     def exceptions_log_path(cls, for_pid=None, in_dir=None):
         """Get the path to either the shared or pid-specific fatal errors log file."""
@@ -294,7 +265,7 @@ def exceptions_log_path(cls, for_pid=None, in_dir=None):
         )
 
     @classmethod
-    def log_exception(cls, msg):
+    def _log_exception(cls, msg):
         """Try to log an error message to this process's error log and the shared error log.
 
         NB: Doesn't raise (logs an error instead).
@@ -446,31 +417,8 @@ def _format_unhandled_exception_log(cls, exc, tb, add_newline, should_print_back
             maybe_newline=maybe_newline,
         )
 
-    _EXIT_FAILURE_TERMINAL_MESSAGE_FORMAT = """\
-{timestamp_msg}{terminal_msg}{details_msg}
-"""
-
     @classmethod
-    def _exit_with_failure(cls, terminal_msg):
-        timestamp_msg = (
-            f"timestamp: {cls._iso_timestamp_for_now()}\n"
-            if cls._should_print_backtrace_to_terminal
-            else ""
-        )
-        details_msg = (
-            ""
-            if cls._should_print_backtrace_to_terminal
-            else "\n\n(Use --print-exception-stacktrace to see more error details.)"
-        )
-        terminal_msg = terminal_msg or "<no exit reason provided>"
-        formatted_terminal_msg = cls._EXIT_FAILURE_TERMINAL_MESSAGE_FORMAT.format(
-            timestamp_msg=timestamp_msg, terminal_msg=terminal_msg, details_msg=details_msg
-        )
-        # Exit with failure, printing a message to the terminal (or whatever the interactive stream is).
-        cls._exiter.exit_and_fail(msg=formatted_terminal_msg, out=cls._interactive_output_stream)
-
-    @classmethod
-    def log_unhandled_exception(cls, exc_class=None, exc=None, tb=None, add_newline=False):
+    def log_exception(cls, exc_class=None, exc=None, tb=None, add_newline=False):
         """Logs an unhandled exception to a variety of locations."""
         exc_class = exc_class or sys.exc_info()[0]
         exc = exc or sys.exc_info()[1]
@@ -486,37 +434,13 @@ def log_unhandled_exception(cls, exc_class=None, exc=None, tb=None, add_newline=
             exception_log_entry = cls._format_unhandled_exception_log(
                 exc, tb, add_newline, should_print_backtrace=True
             )
-            cls.log_exception(exception_log_entry)
+            cls._log_exception(exception_log_entry)
         except Exception as e:
             extra_err_msg = "Additional error logging unhandled exception {}: {}".format(exc, e)
             logger.error(extra_err_msg)
 
-        # Generate an unhandled exception report fit to be printed to the terminal (respecting the
-        # Exiter's should_print_backtrace field).
-        if cls._should_print_backtrace_to_terminal:
-            stderr_printed_error = cls._format_unhandled_exception_log(
-                exc, tb, add_newline, should_print_backtrace=cls._should_print_backtrace_to_terminal
-            )
-            if extra_err_msg:
-                stderr_printed_error = "{}\n{}".format(stderr_printed_error, extra_err_msg)
-        else:
-            # If the user didn't ask for a backtrace, show a succinct error message without
-            # all the exception-related preamble.  A power-user/pants developer can still
-            # get all the preamble info along with the backtrace, but the end user shouldn't
-            # see that boilerplate by default.
-            error_msgs = getattr(exc, "end_user_messages", lambda: [str(exc)])()
-            stderr_printed_error = "\n" + "\n".join(f"ERROR: {msg}" for msg in error_msgs)
-        return stderr_printed_error
-
-    @classmethod
-    def _log_unhandled_exception_and_exit(
-        cls, exc_class=None, exc=None, tb=None, add_newline=False
-    ):
-        """A sys.excepthook implementation which logs the error and exits with failure."""
-        stderr_printed_error = cls.log_unhandled_exception(
-            exc_class=exc_class, exc=exc, tb=tb, add_newline=add_newline
-        )
-        cls._exit_with_failure(stderr_printed_error)
+        # Generate an unhandled exception report fit to be printed to the terminal.
+        logger.exception(exc)
 
     _CATCHABLE_SIGNAL_ERROR_LOG_FORMAT = """\
 Signal {signum} ({signame}) was raised. Exiting with failure.{formatted_traceback}
@@ -524,8 +448,7 @@ def _log_unhandled_exception_and_exit(
 
     @classmethod
     def _handle_signal_gracefully(cls, signum, signame, traceback_lines):
-        """Signal handler for non-fatal signals which raises or logs an error and exits with
-        failure."""
+        """Signal handler for non-fatal signals which raises or logs an error."""
         # Extract the stack, and format an entry to be written to the exception log.
         formatted_traceback = cls._format_traceback(
             traceback_lines=traceback_lines, should_print_backtrace=True
@@ -536,7 +459,7 @@ def _handle_signal_gracefully(cls, signum, signame, traceback_lines):
         # TODO: determine the appropriate signal-safe behavior here (to avoid writing to our file
         # descriptors re-entrantly, which raises an IOError).
         # This method catches any exceptions raised within it.
-        cls.log_exception(signal_error_log_entry)
+        cls._log_exception(signal_error_log_entry)
 
         # Create a potentially-abbreviated traceback for the terminal or other interactive stream.
         formatted_traceback_for_terminal = cls._format_traceback(
@@ -546,19 +469,19 @@ def _handle_signal_gracefully(cls, signum, signame, traceback_lines):
         terminal_log_entry = cls._CATCHABLE_SIGNAL_ERROR_LOG_FORMAT.format(
             signum=signum, signame=signame, formatted_traceback=formatted_traceback_for_terminal
         )
-        # Exit, printing the output to the terminal.
-        cls._exit_with_failure(terminal_log_entry)
+        # Print the output via standard logging.
+        logger.error(terminal_log_entry)
 
 
 # Setup global state such as signal handlers and sys.excepthook with probably-safe values at module
 # import time.
 # Set the log location for writing logs before bootstrap options are parsed.
 ExceptionSink.reset_log_location(os.getcwd())
-# Sets except hook for exceptions at import time.
-ExceptionSink._reset_exiter(Exiter(exiter=sys.exit))
+# Sets exiter at import time.
+ExceptionSink.reset_exiter(sys.exit)
 # Sets a SIGUSR2 handler.
 ExceptionSink.reset_interactive_output_stream(sys.stderr.buffer)
-# Sets a handler that logs nonfatal signals to the exception sink before exiting.
+# Sets a handler that logs nonfatal signals to the exception sink.
 ExceptionSink.reset_signal_handler(SignalHandler())
 # Set whether to print stacktraces on exceptions or signals during import time.
 # NB: This will be overridden by bootstrap options in PantsRunner, so we avoid printing out a full
diff --git a/src/python/pants/base/exiter.py b/src/python/pants/base/exiter.py
index d1a42366c1c3..745f92a1590f 100644
--- a/src/python/pants/base/exiter.py
+++ b/src/python/pants/base/exiter.py
@@ -1,89 +1,8 @@
-# Copyright 2015 Pants project contributors (see CONTRIBUTORS.md).
+# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
 # Licensed under the Apache License, Version 2.0 (see LICENSE).
 
-import logging
-import sys
-import traceback
-from typing import Callable, Union
-
-from pants.util.strutil import ensure_binary
-
-logger = logging.getLogger(__name__)
-
-
-# Centralize integer return codes for the pants process. We just use a single bit for now.
-# TODO: make these into an enum!
-PANTS_SUCCEEDED_EXIT_CODE = 0
-PANTS_FAILED_EXIT_CODE = 1
-
 ExitCode = int
 
-
-class Exiter:
-    """A class that provides standard runtime exit behavior.
-
-    `pants.base.exception_sink.ExceptionSink` handles exceptions and fatal signals, delegating to an
-    Exiter instance which can be set process-globally with ExceptionSink.reset_exiter(). Call
-    .exit() or .exit_and_fail() on an Exiter instance when you wish to exit the runtime.
-    """
-
-    def __init__(self, exiter: Callable[[Union[ExitCode, str, object]], None] = sys.exit):
-        """
-        :param exiter: A function to be called to conduct the final exit of the runtime. (Optional)
-        """
-        # Since we have some exit paths that run via the sys.excepthook,
-        # symbols we use can become garbage collected before we use them; ie:
-        # we can find `sys` and `traceback` are `None`.  As a result we capture
-        # all symbols we need here to ensure we function in excepthook context.
-        # See: http://stackoverflow.com/questions/2572172/referencing-other-modules-in-atexit
-        self._exit = exiter
-
-    def __call__(self, *args, **kwargs):
-        """Map class calls to self.exit() to support sys.exit() fungibility."""
-        return self.exit(*args, **kwargs)
-
-    def exit(self, result=PANTS_SUCCEEDED_EXIT_CODE, msg=None, out=None):
-        """Exits the runtime.
-
-        :param result: The exit status. Typically either PANTS_SUCCEEDED_EXIT_CODE or
-                       PANTS_FAILED_EXIT_CODE, but can be a string as well. (Optional)
-        :param msg: A string message to print to stderr or another custom file desciptor before exiting.
-                    (Optional)
-        :param out: The file descriptor to emit `msg` to. (Optional)
-        """
-        if msg:
-            out = out or sys.stderr
-            if hasattr(out, "buffer"):
-                out = out.buffer
-
-            msg = ensure_binary(msg)
-            try:
-                out.write(msg)
-                out.write(b"\n")
-                # TODO: Determine whether this call is a no-op because the stream gets flushed on exit, or
-                # if we could lose what we just printed, e.g. if we get interrupted by a signal while
-                # exiting and the stream is buffered like stdout.
-                out.flush()
-            except Exception as e:
-                # If the file is already closed, or any other error occurs, just log it and continue to
-                # exit.
-                if msg:
-                    logger.warning(
-                        "Encountered error when trying to log this message: {}, \n "
-                        "exception: {} \n out: {}".format(msg, e, out)
-                    )
-                    # In pantsd, this won't go anywhere, because there's really nowhere for us to log if we
-                    # can't log :(
-                    # Not in pantsd, this will end up in sys.stderr.
-                    traceback.print_stack()
-                logger.exception(e)
-        self._exit(result)
-
-    def exit_and_fail(self, msg=None, out=None):
-        """Exits the runtime with a nonzero exit code, indicating failure.
-
-        :param msg: A string message to print to stderr or another custom file desciptor before exiting.
-                    (Optional)
-        :param out: The file descriptor to emit `msg` to. (Optional)
-        """
-        self.exit(result=PANTS_FAILED_EXIT_CODE, msg=msg, out=out)
+# Centralize integer return codes for the pants process.
+PANTS_SUCCEEDED_EXIT_CODE: ExitCode = 0
+PANTS_FAILED_EXIT_CODE: ExitCode = 1
diff --git a/src/python/pants/bin/daemon_pants_runner.py b/src/python/pants/bin/daemon_pants_runner.py
index 4ca1e880095a..da19d1ed5107 100644
--- a/src/python/pants/bin/daemon_pants_runner.py
+++ b/src/python/pants/bin/daemon_pants_runner.py
@@ -10,12 +10,15 @@
 from typing import Dict, Tuple
 
 from pants.base.exception_sink import ExceptionSink
-from pants.base.exiter import PANTS_FAILED_EXIT_CODE, PANTS_SUCCEEDED_EXIT_CODE
+from pants.base.exiter import PANTS_FAILED_EXIT_CODE, ExitCode
 from pants.bin.local_pants_runner import LocalPantsRunner
-from pants.engine.internals.native import RawFdRunner
-from pants.engine.unions import UnionMembership
-from pants.help.help_printer import HelpPrinter
-from pants.init.specs_calculator import SpecsCalculator
+from pants.engine.internals.native import Native, RawFdRunner
+from pants.init.logging import (
+    clear_logging_handlers,
+    get_logging_handlers,
+    set_logging_handlers,
+    setup_logging,
+)
 from pants.init.util import clean_global_runtime_state
 from pants.java.nailgun_protocol import NailgunProtocol
 from pants.option.options_bootstrapper import OptionsBootstrapper
@@ -127,7 +130,28 @@ def should_keep_polling(now):
                     "Timed out while waiting for another pants invocation to finish."
                 )
 
-    def _run(self, working_dir: str) -> int:
+    @contextmanager
+    def _stderr_logging(self, global_bootstrap_options):
+        """Temporarily replaces existing handlers (ie, the pantsd handler) with a stderr handler.
+
+        In the context of pantsd, there will be an existing handler for the pantsd log, which we
+        temporarily replace. Making them additive would cause per-run logs to go to pantsd, which
+        we don't want.
+
+        TODO: It would be good to handle logging destinations entirely via the threadlocal state
+        rather than via handler mutations.
+        """
+        handlers = get_logging_handlers()
+        try:
+            clear_logging_handlers()
+            Native().override_thread_logging_destination_to_just_stderr()
+            setup_logging(global_bootstrap_options)
+            yield
+        finally:
+            Native().override_thread_logging_destination_to_just_pantsd()
+            set_logging_handlers(handlers)
+
+    def _run(self, working_dir: str) -> ExitCode:
         """Run a single daemonized run of Pants.
 
         All aspects of the `sys` global should already have been replaced in `__call__`, so this
@@ -135,8 +159,12 @@ def _run(self, working_dir: str) -> int:
         environment.
         """
 
-        exit_code = PANTS_SUCCEEDED_EXIT_CODE
         try:
+            # Capture the client's start time, which we propagate here in order to get an accurate
+            # view of total time.
+            env_start_time = os.environ.get("PANTSD_RUNTRACKER_CLIENT_START_TIME", None)
+            start_time = float(env_start_time) if env_start_time else time.time()
+
             if working_dir != os.getcwd():
                 # This is likely an implementation error in the client (which we control).
                 raise Exception(
@@ -145,50 +173,29 @@ def _run(self, working_dir: str) -> int:
                     f"server in: {os.getcwd()}"
                 )
 
-            # Clean global state.
+            # Clear global mutable state before entering `LocalPantsRunner`. Note that we use
+            # `sys.argv` and `os.environ`, since they have been mutated to maintain the illusion
+            # of a local run: once we allow for concurrent runs, this information should be
+            # propagated down from the caller.
+            #   see https://github.com/pantsbuild/pants/issues/7654
             clean_global_runtime_state(reset_subsystem=True)
-
-            options_bootstrapper = OptionsBootstrapper.create(args=sys.argv, env=os.environ)
-            options, build_config = LocalPantsRunner.parse_options(options_bootstrapper)
-
-            global_options = options.for_global_scope()
-            session = self._scheduler_service.prepare_graph(options)
-
-            specs = SpecsCalculator.create(
-                options=options,
-                session=session.scheduler_session,
-                exclude_patterns=tuple(global_options.exclude_target_regexp),
-                tags=tuple(global_options.tag) if global_options.tag else (),
-            )
-
-            if options.help_request:
-                help_printer = HelpPrinter(
-                    options=options, union_membership=UnionMembership(build_config.union_rules()),
+            options_bootstrapper = OptionsBootstrapper.create(env=os.environ, args=sys.argv)
+            bootstrap_options = options_bootstrapper.bootstrap_options
+            global_bootstrap_options = bootstrap_options.for_global_scope()
+
+            # Run using the pre-warmed Session.
+            with self._stderr_logging(global_bootstrap_options):
+                scheduler = self._scheduler_service.prepare()
+                runner = LocalPantsRunner.create(
+                    os.environ, options_bootstrapper, scheduler=scheduler
                 )
-                return help_printer.print_help()
-
-            exit_code = self._scheduler_service.graph_run_v2(
-                session, specs, options, options_bootstrapper
-            )
-            # self.scheduler_service.graph_run_v2 will already run v2 or ambiguous goals. We should
-            # only enter this code path if v1 is set.
-            if global_options.v1:
-                runner = LocalPantsRunner.create(os.environ, options_bootstrapper, specs, session)
-
-                env_start_time = os.environ.get("PANTSD_RUNTRACKER_CLIENT_START_TIME", None)
-                start_time = float(env_start_time) if env_start_time else None
-                runner.set_start_time(start_time)
-                # TODO: This is almost certainly not correct: we should likely have the entire run
-                # occur inside LocalPantsRunner now, and have it return an exit code directly.
-                exit_code = runner.run()
-
-            return exit_code
+                return runner.run(start_time)
 
         except KeyboardInterrupt:
             print("Interrupted by user.\n", file=sys.stderr)
             return PANTS_FAILED_EXIT_CODE
         except Exception as e:
-            ExceptionSink.log_unhandled_exception(exc=e)
+            ExceptionSink.log_exception(exc=e)
             return PANTS_FAILED_EXIT_CODE
 
     def __call__(
@@ -200,7 +207,7 @@ def __call__(
         stdin_fd: int,
         stdout_fd: int,
         stderr_fd: int,
-    ) -> int:
+    ) -> ExitCode:
         request_timeout = float(env.get("PANTSD_REQUEST_TIMEOUT_LIMIT", -1))
         # NB: Order matters: we acquire a lock before mutating either `sys.std*`, `os.environ`, etc.
         with self._one_run_at_a_time(stderr_fd, timeout=request_timeout), self.nailgun_stdio(
@@ -208,4 +215,8 @@ def __call__(
         ), hermetic_environment_as(**env), argv_as((command,) + args):
             # NB: Run implements exception handling, so only the most primitive errors will escape
             # this function, where they will be logged to the pantsd.log by the server.
-            return self._run(working_directory.decode())
+            logger.info(f"handling request: `{' '.join(args)}`")
+            try:
+                return self._run(working_directory.decode())
+            finally:
+                logger.info(f"request completed: `{' '.join(args)}`")
diff --git a/src/python/pants/bin/goal_runner.py b/src/python/pants/bin/goal_runner.py
index 68ceffc6e187..479357c040d8 100644
--- a/src/python/pants/bin/goal_runner.py
+++ b/src/python/pants/bin/goal_runner.py
@@ -2,9 +2,9 @@
 # Licensed under the Apache License, Version 2.0 (see LICENSE).
 
 import logging
-import sys
 from typing import List
 
+from pants.base.exiter import ExitCode
 from pants.base.specs import AddressSpecs, FilesystemSpecs, SingleAddress, Specs
 from pants.base.workunit import WorkUnit, WorkUnitLabel
 from pants.build_graph.build_configuration import BuildConfiguration
@@ -38,7 +38,6 @@ def __init__(
         reporting: Reporting,
         graph_session: LegacyGraphSession,
         specs: Specs,
-        exiter=sys.exit,
     ) -> None:
         """
         :param root_dir: The root directory of the pants workspace (aka the "build root").
@@ -48,7 +47,6 @@ def __init__(
         :param reporting: The global, pre-initialized Reporting instance.
         :param graph_session: The graph session for this run.
         :param specs: The specs for this run, i.e. either the address or filesystem specs.
-        :param func exiter: A function that accepts an exit code value and exits. (for tests, Optional)
         """
         self._root_dir = root_dir
         self._options_bootstrapper = options_bootstrapper
@@ -58,7 +56,6 @@ def __init__(
         self._reporting = reporting
         self._graph_session = graph_session
         self._specs = specs
-        self._exiter = exiter
 
         self._global_options = options.for_global_scope()
         self._fail_fast = self._global_options.fail_fast
@@ -182,7 +179,7 @@ def _is_valid_workdir(self, workdir):
         )
         return False
 
-    def _execute_engine(self) -> int:
+    def _execute_engine(self) -> ExitCode:
         engine = RoundEngine()
         sorted_goal_infos = engine.sort_goals(self._context, self._goals)
         RunTracker.global_instance().set_sorted_goal_infos(sorted_goal_infos)
@@ -193,7 +190,7 @@ def _execute_engine(self) -> int:
 
         return result
 
-    def _run_goals(self) -> int:
+    def _run_goals(self) -> ExitCode:
         should_kill_nailguns = self._kill_nailguns
 
         try:
@@ -221,7 +218,7 @@ def _run_goals(self) -> int:
 
         return result
 
-    def run(self) -> int:
+    def run(self) -> ExitCode:
         global_options = self._context.options.for_global_scope()
 
         if not self._is_valid_workdir(global_options.pants_workdir):
diff --git a/src/python/pants/bin/local_pants_runner.py b/src/python/pants/bin/local_pants_runner.py
index 248983b2382f..c88ecadced90 100644
--- a/src/python/pants/bin/local_pants_runner.py
+++ b/src/python/pants/bin/local_pants_runner.py
@@ -3,27 +3,26 @@
 
 import logging
 import os
-from contextlib import contextmanager
 from dataclasses import dataclass
 from typing import Mapping, Optional, Tuple
 
 from pants.base.build_environment import get_buildroot
 from pants.base.cmd_line_spec_parser import CmdLineSpecParser
 from pants.base.exception_sink import ExceptionSink
-from pants.base.exiter import PANTS_FAILED_EXIT_CODE, PANTS_SUCCEEDED_EXIT_CODE, Exiter
+from pants.base.exiter import PANTS_FAILED_EXIT_CODE, PANTS_SUCCEEDED_EXIT_CODE, ExitCode
 from pants.base.specs import Specs
 from pants.base.workunit import WorkUnit
 from pants.bin.goal_runner import GoalRunner
 from pants.build_graph.build_configuration import BuildConfiguration
 from pants.engine.internals.native import Native
+from pants.engine.internals.scheduler import ExecutionError
 from pants.engine.unions import UnionMembership
 from pants.goal.run_tracker import RunTracker
 from pants.help.help_printer import HelpPrinter
-from pants.init.engine_initializer import EngineInitializer, LegacyGraphSession
-from pants.init.logging import (
-    clear_previous_loggers,
-    setup_logging_to_file,
-    setup_logging_to_stderr,
+from pants.init.engine_initializer import (
+    EngineInitializer,
+    LegacyGraphScheduler,
+    LegacyGraphSession,
 )
 from pants.init.options_initializer import BuildConfigInitializer, OptionsInitializer
 from pants.init.repro import Repro, Reproducer
@@ -35,74 +34,12 @@
 from pants.reporting.streaming_workunit_handler import StreamingWorkunitHandler
 from pants.subsystem.subsystem import Subsystem
 from pants.util.contextutil import maybe_profiled
-from pants.util.logging import LogLevel
 
 logger = logging.getLogger(__name__)
 
 
-class LocalExiter(Exiter):
-    @classmethod
-    @contextmanager
-    def wrap_global_exiter(cls, run_tracker, repro):
-        with ExceptionSink.exiter_as(
-            lambda previous_exiter: cls(run_tracker, repro, previous_exiter)
-        ):
-            yield
-
-    def __init__(self, run_tracker, repro, previous_exiter: Exiter) -> None:
-        self._run_tracker = run_tracker
-        self._repro = repro
-        super().__init__(previous_exiter)
-
-    def exit(self, result=PANTS_SUCCEEDED_EXIT_CODE, msg=None, *args, **kwargs):
-        # These strings are prepended to the existing exit message when calling the superclass .exit().
-        additional_messages = []
-        try:
-            if not self._run_tracker.has_ended():
-                if result == PANTS_SUCCEEDED_EXIT_CODE:
-                    outcome = WorkUnit.SUCCESS
-                elif result == PANTS_FAILED_EXIT_CODE:
-                    outcome = WorkUnit.FAILURE
-                else:
-                    run_tracker_msg = (
-                        "unrecognized exit code {} provided to {}.exit() -- "
-                        "interpreting as a failure in the run tracker".format(
-                            result, type(self).__name__
-                        )
-                    )
-                    # Log the unrecognized exit code to the fatal exception log.
-                    ExceptionSink.log_exception(run_tracker_msg)
-                    # Ensure the unrecognized exit code message is also logged to the terminal.
-                    additional_messages.append(run_tracker_msg)
-                    outcome = WorkUnit.FAILURE
-
-                self._run_tracker.set_root_outcome(outcome)
-                run_tracker_result = self._run_tracker.end()
-                assert (
-                    result == run_tracker_result
-                ), "pants exit code not correctly recorded by run tracker"
-        except ValueError as e:
-            # If we have been interrupted by a signal, calling .end() sometimes writes to a closed file,
-            # so we just log that fact here and keep going.
-            exception_string = str(e)
-            ExceptionSink.log_exception(exception_string)
-            additional_messages.append(exception_string)
-        finally:
-            if self._repro:
-                # TODO: Have Repro capture the 'after' state (as a diff) as well? (in reference to the below
-                # 'before' state comment)
-                # NB: this writes to the logger, which is expected to still be alive if we are exiting from
-                # a signal.
-                self._repro.log_location_of_repro_file()
-
-        if additional_messages:
-            msg = "{}\n\n{}".format("\n".join(additional_messages), msg or "")
-
-        super().exit(result=result, msg=msg, *args, **kwargs)
-
-
 @dataclass
-class LocalPantsRunner(ExceptionSink.AccessGlobalExiterMixin):
+class LocalPantsRunner:
     """Handles a single pants invocation running in the process-local context.
 
     build_root: The build root for this run.
@@ -111,7 +48,6 @@ class LocalPantsRunner(ExceptionSink.AccessGlobalExiterMixin):
     build_config: The parsed build configuration for this run.
     specs: The specs for this run, i.e. either the address or filesystem specs.
     graph_session: A LegacyGraphSession instance for graph reuse.
-    is_daemon: Whether or not this run was launched with a daemon graph helper.
     profile_path: The profile path - if any (from from the `PANTS_PROFILE` env var).
     """
 
@@ -122,9 +58,8 @@ class LocalPantsRunner(ExceptionSink.AccessGlobalExiterMixin):
     specs: Specs
     graph_session: LegacyGraphSession
     union_membership: UnionMembership
-    is_daemon: bool
     profile_path: Optional[str]
-    _run_tracker: Optional[RunTracker] = None
+    _run_tracker: RunTracker
     _reporting: Optional[Reporting] = None
     _repro: Optional[Repro] = None
 
@@ -141,10 +76,11 @@ def _init_graph_session(
         options_bootstrapper: OptionsBootstrapper,
         build_config: BuildConfiguration,
         options: Options,
+        scheduler: Optional[LegacyGraphScheduler] = None,
     ) -> LegacyGraphSession:
         native = Native()
         native.set_panic_handler()
-        graph_scheduler_helper = EngineInitializer.setup_legacy_graph(
+        graph_scheduler_helper = scheduler or EngineInitializer.setup_legacy_graph(
             native, options_bootstrapper, build_config
         )
 
@@ -170,29 +106,19 @@ def create(
         cls,
         env: Mapping[str, str],
         options_bootstrapper: OptionsBootstrapper,
-        specs: Optional[Specs] = None,
-        daemon_graph_session: Optional[LegacyGraphSession] = None,
+        scheduler: Optional[LegacyGraphScheduler] = None,
     ) -> "LocalPantsRunner":
         """Creates a new LocalPantsRunner instance by parsing options.
 
+        By the time this method runs, logging will already have been initialized in either
+        PantsRunner or DaemonPantsRunner.
+
         :param env: The environment (e.g. os.environ) for this run.
         :param options_bootstrapper: The OptionsBootstrapper instance to reuse.
-        :param specs: The specs for this run, i.e. either the address or filesystem specs.
-        :param daemon_graph_session: The graph helper for this session.
+        :param scheduler: If being called from the daemon, a warmed scheduler to use.
         """
         build_root = get_buildroot()
-        global_options = options_bootstrapper.bootstrap_options.for_global_scope()
-        # This works as expected due to the encapsulated_logger in DaemonPantsRunner and
-        # we don't have to gate logging setup anymore.
-
-        level = LogLevel.ERROR if getattr(global_options, "quiet", False) else global_options.level
-        ignores = global_options.ignore_pants_warnings
-        clear_previous_loggers()
-        setup_logging_to_stderr(level, warnings_filter_regexes=ignores)
-        log_dir = global_options.logdir
-        if log_dir:
-            setup_logging_to_file(level, log_dir=log_dir, warnings_filter_regexes=ignores)
-
+        global_bootstrap_options = options_bootstrapper.bootstrap_options.for_global_scope()
         options, build_config = LocalPantsRunner.parse_options(options_bootstrapper)
 
         # Option values are usually computed lazily on demand,
@@ -201,28 +127,25 @@ def create(
             options.for_scope(scope)
 
         # Verify configs.
-        if global_options.verify_config:
+        if global_bootstrap_options.verify_config:
             options.verify_configs(options_bootstrapper.config)
 
         union_membership = UnionMembership(build_config.union_rules())
 
-        # If we're running with the daemon, we'll be handed a session from the
-        # resident graph helper - otherwise initialize a new one here.
-        graph_session = (
-            daemon_graph_session
-            if daemon_graph_session
-            else cls._init_graph_session(options_bootstrapper, build_config, options)
+        # If we're running with the daemon, we'll be handed a warmed Scheduler, which we use
+        # to initialize a session here.
+        graph_session = cls._init_graph_session(
+            options_bootstrapper, build_config, options, scheduler
         )
 
-        if specs is None:
-            global_options = options.for_global_scope()
-            specs = SpecsCalculator.create(
-                options=options,
-                build_root=build_root,
-                session=graph_session.scheduler_session,
-                exclude_patterns=tuple(global_options.exclude_target_regexp),
-                tags=tuple(global_options.tag),
-            )
+        global_options = options.for_global_scope()
+        specs = SpecsCalculator.create(
+            options=options,
+            build_root=build_root,
+            session=graph_session.scheduler_session,
+            exclude_patterns=tuple(global_options.exclude_target_regexp),
+            tags=tuple(global_options.tag),
+        )
 
         profile_path = env.get("PANTS_PROFILE")
 
@@ -234,14 +157,11 @@ def create(
             specs=specs,
             graph_session=graph_session,
             union_membership=union_membership,
-            is_daemon=daemon_graph_session is not None,
             profile_path=profile_path,
+            _run_tracker=RunTracker.global_instance(),
         )
 
-    def set_start_time(self, start_time: Optional[float]) -> None:
-        # Launch RunTracker as early as possible (before .run() is called).
-        self._run_tracker = RunTracker.global_instance()
-
+    def _set_start_time(self, start_time: float) -> None:
         # Propagates parent_build_id to pants runs that may be called from this pants run.
         os.environ["PANTS_PARENT_BUILD_ID"] = self._run_tracker.run_id
 
@@ -258,7 +178,7 @@ def set_start_time(self, start_time: Optional[float]) -> None:
         if self._repro:
             self._repro.capture(self._run_tracker.run_info.get_as_dict())
 
-    def _maybe_run_v1(self, v1: bool) -> int:
+    def _maybe_run_v1(self, v1: bool) -> ExitCode:
         v1_goals, ambiguous_goals, _ = self.options.goals_by_version
         if not v1:
             if v1_goals:
@@ -280,43 +200,57 @@ def _maybe_run_v1(self, v1: bool) -> int:
                 self.options_bootstrapper,
                 self.options,
                 self.build_config,
-                self._run_tracker,  # type: ignore
+                self._run_tracker,
                 self._reporting,  # type: ignore
                 self.graph_session,
                 self.specs,
-                self._exiter,
             )
             .create()
             .run()
         )
 
-    def _maybe_run_v2(self, v2: bool) -> int:
-        # N.B. For daemon runs, @goal_rules are invoked pre-fork -
-        # so this path only serves the non-daemon run mode.
-        if self.is_daemon:
-            return PANTS_SUCCEEDED_EXIT_CODE
-
+    def _maybe_run_v2(self, v2: bool) -> ExitCode:
         _, ambiguous_goals, v2_goals = self.options.goals_by_version
         goals = v2_goals + (ambiguous_goals if v2 else tuple())
         if self._run_tracker:
             self._run_tracker.set_v2_goal_rule_names(goals)
         if not goals:
             return PANTS_SUCCEEDED_EXIT_CODE
-
+        global_options = self.options.for_global_scope()
+        if not global_options.get("loop", False):
+            return self._maybe_run_v2_body(goals, poll=False)
+
+        iterations = global_options.loop_max
+        exit_code = PANTS_SUCCEEDED_EXIT_CODE
+        while iterations:
+            # NB: We generate a new "run id" per iteration of the loop in order to allow us to
+            # observe fresh values for Goals. See notes in `scheduler.rs`.
+            self.graph_session.scheduler_session.new_run_id()
+            try:
+                exit_code = self._maybe_run_v2_body(goals, poll=True)
+            except ExecutionError as e:
+                logger.warning(e)
+            iterations -= 1
+
+        return exit_code
+
+    def _maybe_run_v2_body(self, goals, poll: bool) -> ExitCode:
         return self.graph_session.run_goal_rules(
             options_bootstrapper=self.options_bootstrapper,
             union_membership=self.union_membership,
             options=self.options,
             goals=goals,
             specs=self.specs,
+            poll=poll,
+            poll_delay=(0.1 if poll else None),
         )
 
     @staticmethod
-    def _compute_final_exit_code(*codes):
+    def _merge_exit_codes(code: ExitCode, *codes: ExitCode) -> ExitCode:
         """Returns the exit code with higher abs value in case of negative values."""
-        max_code = None
+        max_code = code
         for code in codes:
-            if max_code is None or abs(max_code) < abs(code):
+            if abs(max_code) < abs(code):
                 max_code = code
         return max_code
 
@@ -328,13 +262,62 @@ def _update_stats(self):
         if engine_workunits:
             self._run_tracker.report.bulk_record_workunits(engine_workunits)
 
-    def run(self):
-        global_options = self.options.for_global_scope()
+    def _finish_run(self, code: ExitCode) -> ExitCode:
+        """Checks that the RunTracker is in good shape to exit, and then returns its exit code.
 
-        exiter = LocalExiter.wrap_global_exiter(self._run_tracker, self._repro)
-        profiled = maybe_profiled(self.profile_path)
+        TODO: The RunTracker's exit code will likely not be relevant in v2: the exit codes of
+        individual `@goal_rule`s are everything in that case.
+        """
+
+        run_tracker_result = PANTS_SUCCEEDED_EXIT_CODE
+
+        # These strings are prepended to the existing exit message when calling the superclass .exit().
+        additional_messages = []
+        try:
+            self._update_stats()
+
+            if code == PANTS_SUCCEEDED_EXIT_CODE:
+                outcome = WorkUnit.SUCCESS
+            elif code == PANTS_FAILED_EXIT_CODE:
+                outcome = WorkUnit.FAILURE
+            else:
+                run_tracker_msg = (
+                    "unrecognized exit code {} provided to {}.exit() -- "
+                    "interpreting as a failure in the run tracker".format(code, type(self).__name__)
+                )
+                # Log the unrecognized exit code to the fatal exception log.
+                ExceptionSink.log_exception(exc=Exception(run_tracker_msg))
+                # Ensure the unrecognized exit code message is also logged to the terminal.
+                additional_messages.append(run_tracker_msg)
+                outcome = WorkUnit.FAILURE
+
+            self._run_tracker.set_root_outcome(outcome)
+            run_tracker_result = self._run_tracker.end()
+        except ValueError as e:
+            # If we have been interrupted by a signal, calling .end() sometimes writes to a closed file,
+            # so we just log that fact here and keep going.
+            ExceptionSink.log_exception(exc=e)
+        finally:
+            if self._repro:
+                # TODO: Have Repro capture the 'after' state (as a diff) as well? (in reference to the below
+                # 'before' state comment)
+                # NB: this writes to the logger, which is expected to still be alive if we are exiting from
+                # a signal.
+                self._repro.log_location_of_repro_file()
+
+        if additional_messages:
+            # NB: We do not log to the exceptions log in this case, because we expect that these are
+            # higher level unstructed errors: strutured versions will already have been written at
+            # various places.
+            logger.error("\n".join(additional_messages))
+
+        return run_tracker_result
+
+    def run(self, start_time: float) -> ExitCode:
+        self._set_start_time(start_time)
 
-        with exiter, profiled:
+        with maybe_profiled(self.profile_path):
+            global_options = self.options.for_global_scope()
             streaming_handlers = global_options.streaming_workunits_handlers
             report_interval = global_options.streaming_workunits_report_interval
             callbacks = Subsystem.get_streaming_workunit_callbacks(streaming_handlers)
@@ -348,27 +331,18 @@ def run(self):
                 help_printer = HelpPrinter(
                     options=self.options, union_membership=self.union_membership
                 )
-                help_output = help_printer.print_help()
-                self._exiter.exit(help_output)
+                return help_printer.print_help()
 
             v1 = global_options.v1
             v2 = global_options.v2
             with streaming_reporter.session():
+                engine_result, goal_runner_result = PANTS_FAILED_EXIT_CODE, PANTS_FAILED_EXIT_CODE
                 try:
                     engine_result = self._maybe_run_v2(v2)
                     goal_runner_result = self._maybe_run_v1(v1)
-                finally:
-                    run_tracker_result = self._finish_run()
-            final_exit_code = self._compute_final_exit_code(
-                engine_result, goal_runner_result, run_tracker_result
-            )
-            self._exiter.exit(final_exit_code)
-
-    def _finish_run(self):
-        try:
-            self._update_stats()
-            return self._run_tracker.end()
-        except ValueError as e:
-            # Calling .end() sometimes writes to a closed file, so we return a dummy result here.
-            logger.exception(e)
-            return PANTS_SUCCEEDED_EXIT_CODE
+                except Exception as e:
+                    ExceptionSink.log_exception(e)
+                run_tracker_result = self._finish_run(
+                    self._merge_exit_codes(engine_result, goal_runner_result)
+                )
+            return self._merge_exit_codes(engine_result, goal_runner_result, run_tracker_result)
diff --git a/src/python/pants/bin/pants_exe.py b/src/python/pants/bin/pants_exe.py
index adb2f056a786..bf19bcab0756 100644
--- a/src/python/pants/bin/pants_exe.py
+++ b/src/python/pants/bin/pants_exe.py
@@ -6,6 +6,7 @@
 import time
 
 from pants.base.exception_sink import ExceptionSink
+from pants.base.exiter import PANTS_FAILED_EXIT_CODE
 from pants.bin.pants_runner import PantsRunner
 from pants.util.contextutil import maybe_profiled
 
@@ -30,6 +31,8 @@ def main():
         start_time = time.time()
         try:
             runner = PantsRunner(args=sys.argv, env=os.environ)
-            runner.run(start_time)
+            exit_code = runner.run(start_time)
         except KeyboardInterrupt as e:
-            ExceptionSink.get_global_exiter().exit_and_fail("Interrupted by user:\n{}".format(e))
+            print("Interrupted by user:\n{}".format(e), file=sys.stderr)
+            exit_code = PANTS_FAILED_EXIT_CODE
+    ExceptionSink.exit(exit_code)
diff --git a/src/python/pants/bin/pants_runner.py b/src/python/pants/bin/pants_runner.py
index baf20e9f1871..9d571c737bf4 100644
--- a/src/python/pants/bin/pants_runner.py
+++ b/src/python/pants/bin/pants_runner.py
@@ -7,8 +7,9 @@
 from typing import List, Mapping
 
 from pants.base.exception_sink import ExceptionSink
+from pants.base.exiter import ExitCode
 from pants.bin.remote_pants_runner import RemotePantsRunner
-from pants.init.logging import init_rust_logger, setup_logging_to_stderr
+from pants.init.logging import setup_logging
 from pants.init.util import init_workdir
 from pants.option.option_value_container import OptionValueContainer
 from pants.option.options_bootstrapper import OptionsBootstrapper
@@ -17,7 +18,7 @@
 
 
 @dataclass(frozen=True)
-class PantsRunner(ExceptionSink.AccessGlobalExiterMixin):
+class PantsRunner:
     """A higher-level runner that delegates runs to either a LocalPantsRunner or
     RemotePantsRunner."""
 
@@ -31,12 +32,6 @@ def will_terminate_pantsd(self) -> bool:
         _DAEMON_KILLING_GOALS = frozenset(["kill-pantsd", "clean-all"])
         return not frozenset(self.args).isdisjoint(_DAEMON_KILLING_GOALS)
 
-    @staticmethod
-    def _enable_rust_logging(global_bootstrap_options: OptionValueContainer) -> None:
-        log_level = global_bootstrap_options.level
-        init_rust_logger(log_level, global_bootstrap_options.log_show_rust_3rdparty)
-        setup_logging_to_stderr(log_level)
-
     def _should_run_with_pantsd(self, global_bootstrap_options: OptionValueContainer) -> bool:
         # The parent_build_id option is set only for pants commands (inner runs)
         # that were called by other pants command.
@@ -67,10 +62,9 @@ def scrub_pythonpath() -> None:
         if pythonpath and not os.environ.pop("RUNNING_PANTS_FROM_SOURCES", None):
             logger.warning(f"Scrubbed PYTHONPATH={pythonpath} from the environment.")
 
-    def run(self, start_time: float) -> None:
+    def run(self, start_time: float) -> ExitCode:
         self.scrub_pythonpath()
 
-        # TODO could options-bootstrapper be parsed in the runners?
         options_bootstrapper = OptionsBootstrapper.create(env=self.env, args=self.args)
         bootstrap_options = options_bootstrapper.bootstrap_options
         global_bootstrap_options = bootstrap_options.for_global_scope()
@@ -79,9 +73,9 @@ def run(self, start_time: float) -> None:
         workdir_src = init_workdir(global_bootstrap_options)
         ExceptionSink.reset_log_location(workdir_src)
 
-        # We enable Rust logging here,
-        # and everything before it will be routed through regular Python logging.
-        self._enable_rust_logging(global_bootstrap_options)
+        # We enable logging here, and everything before it will be routed through regular
+        # Python logging.
+        setup_logging(global_bootstrap_options)
 
         ExceptionSink.reset_should_print_backtrace_to_terminal(
             global_bootstrap_options.print_exception_stacktrace
@@ -89,8 +83,7 @@ def run(self, start_time: float) -> None:
 
         if self._should_run_with_pantsd(global_bootstrap_options):
             try:
-                RemotePantsRunner(self._exiter, self.args, self.env, options_bootstrapper).run()
-                return
+                return RemotePantsRunner(self.args, self.env, options_bootstrapper).run(start_time)
             except RemotePantsRunner.Fallback as e:
                 logger.warning("Client exception: {!r}, falling back to non-daemon mode".format(e))
 
@@ -98,5 +91,4 @@ def run(self, start_time: float) -> None:
         from pants.bin.local_pants_runner import LocalPantsRunner
 
         runner = LocalPantsRunner.create(env=self.env, options_bootstrapper=options_bootstrapper)
-        runner.set_start_time(start_time)
-        runner.run()
+        return runner.run(start_time)
diff --git a/src/python/pants/bin/remote_pants_runner.py b/src/python/pants/bin/remote_pants_runner.py
index 859ff8d6d980..98a8b8a39a81 100644
--- a/src/python/pants/bin/remote_pants_runner.py
+++ b/src/python/pants/bin/remote_pants_runner.py
@@ -8,6 +8,7 @@
 from typing import List, Mapping
 
 from pants.base.exception_sink import ExceptionSink, SignalHandler
+from pants.base.exiter import ExitCode
 from pants.console.stty_utils import STTYSettings
 from pants.java.nailgun_client import NailgunClient
 from pants.java.nailgun_protocol import NailgunProtocol
@@ -58,7 +59,6 @@ class Fallback(Exception):
     class Terminated(Exception):
         """Raised when an active run is terminated mid-flight."""
 
-    PANTS_COMMAND = "pants"
     RECOVERABLE_EXCEPTIONS = (
         NailgunClient.NailgunConnectionError,
         NailgunClient.NailgunExecutionError,
@@ -66,7 +66,6 @@ class Terminated(Exception):
 
     def __init__(
         self,
-        exiter,
         args: List[str],
         env: Mapping[str, str],
         options_bootstrapper: OptionsBootstrapper,
@@ -75,7 +74,6 @@ def __init__(
         stderr=None,
     ) -> None:
         """
-        :param Exiter exiter: The Exiter instance to use for this run.
         :param args: The arguments (e.g. sys.argv) for this run.
         :param env: The environment (e.g. os.environ) for this run.
         :param options_bootstrapper: The bootstrap options.
@@ -84,7 +82,6 @@ def __init__(
         :param file stderr: The stream representing stderr.
         """
         self._start_time = time.time()
-        self._exiter = exiter
         self._args = args
         self._env = env
         self._options_bootstrapper = options_bootstrapper
@@ -109,7 +106,9 @@ def _backoff(attempt):
         """Minimal backoff strategy for daemon restarts."""
         time.sleep(attempt + (attempt - 1))
 
-    def _run_pants_with_retry(self, pantsd_handle: PantsDaemon.Handle, retries: int = 3):
+    def _run_pants_with_retry(
+        self, pantsd_handle: PantsDaemon.Handle, retries: int = 3
+    ) -> ExitCode:
         """Runs pants remotely with retry and recovery for nascent executions.
 
         :param pantsd_handle: A Handle for the daemon to connect to.
@@ -147,7 +146,7 @@ def _run_pants_with_retry(self, pantsd_handle: PantsDaemon.Handle, retries: int
                 traceback = sys.exc_info()[2]
                 raise self._extract_remote_exception(pantsd_handle.pid, e).with_traceback(traceback)
 
-    def _connect_and_execute(self, pantsd_handle: PantsDaemon.Handle):
+    def _connect_and_execute(self, pantsd_handle: PantsDaemon.Handle) -> ExitCode:
         port = pantsd_handle.port
         pid = pantsd_handle.pid
         # Merge the nailgun TTY capability environment variables with the passed environment dict.
@@ -178,10 +177,7 @@ def _connect_and_execute(self, pantsd_handle: PantsDaemon.Handle):
 
         with self._trapped_signals(client, pantsd_handle.pid), STTYSettings.preserved():
             # Execute the command on the pailgun.
-            result = client.execute(self.PANTS_COMMAND, *self._args, **modified_env)
-
-        # Exit.
-        self._exiter.exit(result)
+            return client.execute(self._args[0], self._args[1:], modified_env)
 
     def _extract_remote_exception(self, pantsd_pid, nailgun_error):
         """Given a NailgunError, returns a Terminated exception with additional info (where
@@ -211,6 +207,6 @@ def _extract_remote_exception(self, pantsd_pid, nailgun_error):
     def _restart_pantsd(self):
         return PantsDaemon.Factory.restart(options_bootstrapper=self._options_bootstrapper)
 
-    def run(self, args=None) -> None:
+    def run(self, start_time: float) -> ExitCode:
         handle = PantsDaemon.Factory.maybe_launch(options_bootstrapper=self._options_bootstrapper)
-        self._run_pants_with_retry(handle)
+        return self._run_pants_with_retry(handle)
diff --git a/src/python/pants/engine/internals/native.py b/src/python/pants/engine/internals/native.py
index cff563708b67..f22e33e3c771 100644
--- a/src/python/pants/engine/internals/native.py
+++ b/src/python/pants/engine/internals/native.py
@@ -16,6 +16,7 @@
 import pkg_resources
 from typing_extensions import Protocol
 
+from pants.base.exiter import ExitCode
 from pants.base.project_tree import Dir, File, Link
 from pants.engine.addresses import Address
 from pants.engine.fs import (
@@ -616,7 +617,7 @@ def __call__(
         stdin_fd: int,
         stdout_fd: int,
         stderr_fd: int,
-    ) -> int:
+    ) -> ExitCode:
         ...
 
 
diff --git a/src/python/pants/goal/run_tracker.py b/src/python/pants/goal/run_tracker.py
index 1327071ec037..d50b5183d706 100644
--- a/src/python/pants/goal/run_tracker.py
+++ b/src/python/pants/goal/run_tracker.py
@@ -17,7 +17,7 @@
 import requests
 
 from pants.auth.basic_auth import BasicAuth
-from pants.base.exiter import PANTS_FAILED_EXIT_CODE, PANTS_SUCCEEDED_EXIT_CODE
+from pants.base.exiter import PANTS_FAILED_EXIT_CODE, PANTS_SUCCEEDED_EXIT_CODE, ExitCode
 from pants.base.run_info import RunInfo
 from pants.base.worker_pool import SubprocPool, WorkerPool
 from pants.base.workunit import WorkUnit, WorkUnitLabel
@@ -213,7 +213,7 @@ def __init__(self, *args, **kwargs):
         # }
         self._target_to_data = {}
 
-        self._end_memoized_result = None
+        self._end_memoized_result: Optional[ExitCode] = None
 
     def set_sorted_goal_infos(self, sorted_goal_infos):
         self._sorted_goal_infos = sorted_goal_infos
@@ -557,7 +557,7 @@ def store_stats(self):
     def has_ended(self) -> bool:
         return self._end_memoized_result is not None
 
-    def end(self):
+    def end(self) -> ExitCode:
         """This pants run is over, so stop tracking it.
 
         Note: If end() has been called once, subsequent calls are no-ops.
diff --git a/src/python/pants/init/logging.py b/src/python/pants/init/logging.py
index e5a0296228f4..9ab72f55c8df 100644
--- a/src/python/pants/init/logging.py
+++ b/src/python/pants/init/logging.py
@@ -6,8 +6,8 @@
 import os
 import sys
 import warnings
-from logging import LogRecord, StreamHandler
-from typing import List, Optional, TextIO
+from logging import Formatter, Handler, LogRecord, StreamHandler
+from typing import List, Optional, TextIO, Tuple
 
 import pants.util.logging as pants_logging
 from pants.base.exception_sink import ExceptionSink
@@ -23,8 +23,7 @@
 
 
 def init_rust_logger(log_level: LogLevel, log_show_rust_3rdparty: bool) -> None:
-    native = Native()
-    native.init_rust_logging(log_level.level, log_show_rust_3rdparty)
+    Native().init_rust_logging(log_level.level, log_show_rust_3rdparty)
 
 
 class NativeHandler(StreamHandler):
@@ -68,12 +67,33 @@ def __repr__(self) -> str:
         )
 
 
-def clear_previous_loggers() -> None:
+class ExceptionFormatter(Formatter):
+    """Uses the `--print-exception-stacktrace` option to decide whether to render stacktraces."""
+
+    def formatException(self, exc_info):
+        if ExceptionSink.should_print_exception_stacktrace:
+            return super().formatException(exc_info)
+        return "\n(Use --print-exception-stacktrace to see more error details.)"
+
+
+def clear_logging_handlers():
     logger = logging.getLogger(None)
-    for handler in logger.handlers:
+    for handler in get_logging_handlers():
         logger.removeHandler(handler)
 
 
+def get_logging_handlers() -> Tuple[Handler, ...]:
+    logger = logging.getLogger(None)
+    return tuple(logger.handlers)
+
+
+def set_logging_handlers(handlers: Tuple[Handler, ...]):
+    clear_logging_handlers()
+    logger = logging.getLogger(None)
+    for handler in handlers:
+        logger.addHandler(handler)
+
+
 def _common_logging_setup(level: LogLevel, warnings_filter_regexes: Optional[List[str]]) -> None:
     def trace_fn(self, message, *args, **kwargs):
         if self.isEnabledFor(LogLevel.TRACE.level):
@@ -96,6 +116,27 @@ def trace_fn(self, message, *args, **kwargs):
         requests_logger.propagate = True
 
 
+def setup_logging(global_bootstrap_options):
+    """Sets up logging for a pants run.
+
+    This is called in two contexts: 1) PantsRunner, 2) DaemonPantsRunner. In the latter case, the
+    loggers are saved and restored around this call, so in both cases it runs with no handlers
+    configured (and asserts so!).
+    """
+    if get_logging_handlers():
+        raise AssertionError("setup_logging should not be called while Handlers are installed.")
+
+    ignores = global_bootstrap_options.ignore_pants_warnings
+    global_level = global_bootstrap_options.level
+    level = LogLevel.ERROR if getattr(global_bootstrap_options, "quiet", False) else global_level
+    log_dir = global_bootstrap_options.logdir
+
+    Native().init_rust_logging(level.level, global_bootstrap_options.log_show_rust_3rdparty)
+    setup_logging_to_stderr(level, warnings_filter_regexes=ignores)
+    if log_dir:
+        setup_logging_to_file(global_level, log_dir=log_dir, warnings_filter_regexes=ignores)
+
+
 def setup_logging_to_stderr(
     level: LogLevel, *, warnings_filter_regexes: Optional[List[str]] = None
 ) -> None:
@@ -108,6 +149,7 @@ def setup_logging_to_stderr(
 
     python_logger = logging.getLogger(None)
     handler = NativeHandler(level, stream=sys.stderr)
+    handler.setFormatter(ExceptionFormatter())
     python_logger.addHandler(handler)
     LogLevel.TRACE.set_level_for(python_logger)
 
@@ -129,7 +171,7 @@ def setup_logging_to_file(
 
     fd = native.setup_pantsd_logger(log_path, level.level)
     ExceptionSink.reset_interactive_output_stream(os.fdopen(os.dup(fd), "a"))
-    native_handler = NativeHandler(level, native_filename=log_path)
+    handler = NativeHandler(level, native_filename=log_path)
 
-    logger.addHandler(native_handler)
-    return native_handler
+    logger.addHandler(handler)
+    return handler
diff --git a/src/python/pants/java/nailgun_client.py b/src/python/pants/java/nailgun_client.py
index 5824d29dccd0..2380e977e31d 100644
--- a/src/python/pants/java/nailgun_client.py
+++ b/src/python/pants/java/nailgun_client.py
@@ -7,7 +7,9 @@
 import socket
 import sys
 import time
+from typing import Dict, List, Optional, cast
 
+from pants.base.exiter import ExitCode
 from pants.java.nailgun_io import NailgunStreamWriter
 from pants.java.nailgun_protocol import ChunkType, MaybeShutdownSocket, NailgunProtocol
 from pants.util.osutil import safe_kill
@@ -250,17 +252,23 @@ def maybe_send_signal(self, signum):
         if self.remote_pid is not None:
             safe_kill(self.remote_pid, signum)
 
-    def execute(self, main_class, cwd=None, *args, **environment):
+    def execute(
+        self,
+        main_class: str,
+        args: List[str],
+        env: Optional[Dict[str, str]] = None,
+        cwd: Optional[str] = None,
+    ) -> ExitCode:
         """Executes the given main_class with any supplied args in the given environment.
 
-        :param string main_class: the fully qualified class name of the main entrypoint
-        :param string cwd: Set the working directory for this command
-        :param list args: any arguments to pass to the main entrypoint
-        :param dict environment: an env mapping made available to native nails via the nail context
+        :param main_class: the fully qualified class name of the main entrypoint
+        :param args: any arguments to pass to the main entrypoint
+        :param env: an env mapping made available to native nails via the nail context
+        :param cwd: Set the working directory for this command
         :returns: the exit code of the main_class.
         :raises: :class:`NailgunClient.NailgunError` if there was an error during execution.
         """
-        environment = dict(**environment)
+        environment = dict(env or {})
         environment.update(self.ENV_DEFAULTS)
         cwd = cwd or os.getcwd()
 
@@ -277,7 +285,8 @@ def execute(self, main_class, cwd=None, *args, **environment):
             exit_on_broken_pipe=self._exit_on_broken_pipe,
         )
         try:
-            return self._session.execute(cwd, main_class, *args, **environment)
+            exit_code = self._session.execute(cwd, main_class, *args, **environment)
+            return cast(ExitCode, exit_code)
         except (socket.error, NailgunProtocol.ProtocolError) as e:
             raise self.NailgunError(
                 address=self._address_string, wrapped_exc=e,
diff --git a/src/python/pants/java/nailgun_executor.py b/src/python/pants/java/nailgun_executor.py
index 9417a5e4b27c..876d05f68e4c 100644
--- a/src/python/pants/java/nailgun_executor.py
+++ b/src/python/pants/java/nailgun_executor.py
@@ -166,7 +166,7 @@ def run(this, stdout=None, stderr=None, stdin=None, cwd=None):
                     logger.debug(
                         "Executing via {ng_desc}: {cmd}".format(ng_desc=nailgun, cmd=this.cmd)
                     )
-                    return nailgun.execute(main, cwd, *args)
+                    return nailgun.execute(main, args, cwd=cwd)
                 except (NailgunClient.NailgunError, self.InitialNailgunConnectTimedOut) as e:
                     self.terminate()
                     raise self.Error(
diff --git a/src/python/pants/java/nailgun_protocol.py b/src/python/pants/java/nailgun_protocol.py
index 01ac9becc2a3..ebf4e0032e88 100644
--- a/src/python/pants/java/nailgun_protocol.py
+++ b/src/python/pants/java/nailgun_protocol.py
@@ -362,6 +362,7 @@ def gen_env_vars():
             for fd_id, fd in zip(STDIO_DESCRIPTORS, (stdin, stdout, stderr)):
                 if fd.isatty():
                     yield (cls.TTY_PATH_ENV.format(fd_id), os.ttyname(fd.fileno()) or b"")
+
         return dict(gen_env_vars())
 
     @classmethod
diff --git a/src/python/pants/option/global_options.py b/src/python/pants/option/global_options.py
index 881deefd7b3f..c61865d66cab 100644
--- a/src/python/pants/option/global_options.py
+++ b/src/python/pants/option/global_options.py
@@ -1072,13 +1072,6 @@ def validate_instance(cls, opts):
 
         Raises pants.option.errors.OptionsError on validation failure.
         """
-        if opts.get("loop") and not opts.enable_pantsd:
-            # TODO: This remains the case today because there are two spots that
-            # call `run_goal_rules`: fixing in a followup.
-            raise OptionsError(
-                "The `--loop` option requires `--enable-pantsd`, in order to watch files."
-            )
-
         if opts.remote_execution and not opts.remote_execution_server:
             raise OptionsError(
                 "The `--remote-execution` option requires also setting "
diff --git a/src/python/pants/pantsd/pants_daemon.py b/src/python/pants/pantsd/pants_daemon.py
index b22b2e46fbd9..7a5b7fbdd618 100644
--- a/src/python/pants/pantsd/pants_daemon.py
+++ b/src/python/pants/pantsd/pants_daemon.py
@@ -13,17 +13,11 @@
 
 from pants.base.build_environment import get_buildroot
 from pants.base.exception_sink import ExceptionSink, SignalHandler
-from pants.base.exiter import Exiter
 from pants.bin.daemon_pants_runner import DaemonPantsRunner
 from pants.engine.internals.native import Native
 from pants.engine.unions import UnionMembership
 from pants.init.engine_initializer import EngineInitializer
-from pants.init.logging import (
-    clear_previous_loggers,
-    init_rust_logger,
-    setup_logging_to_file,
-    setup_logging_to_stderr,
-)
+from pants.init.logging import clear_logging_handlers, init_rust_logger, setup_logging_to_file
 from pants.init.options_initializer import BuildConfigInitializer, OptionsInitializer
 from pants.option.option_value_container import OptionValueContainer
 from pants.option.options_bootstrapper import OptionsBootstrapper
@@ -359,8 +353,7 @@ def _pantsd_logging(self) -> Iterator[IO[str]]:
 
             level = self._log_level
             ignores = bootstrap_options.for_global_scope().ignore_pants_warnings
-            clear_previous_loggers()
-            setup_logging_to_stderr(level, warnings_filter_regexes=ignores)
+            clear_logging_handlers()
             log_dir = os.path.join(self._work_dir, self.name)
             log_handler = setup_logging_to_file(
                 level, log_dir=log_dir, log_filename=self.LOG_NAME, warnings_filter_regexes=ignores
@@ -483,9 +476,8 @@ def run_sync(self):
         # Switch log output to the daemon's log stream from here forward.
         # Also, register an exiter using os._exit to ensure we only close stdio streams once.
         self._close_stdio()
-        with self._pantsd_logging() as log_stream, ExceptionSink.exiter_as(
-            lambda _: Exiter(exiter=os._exit)
-        ):
+        ExceptionSink.reset_exiter(os._exit)
+        with self._pantsd_logging() as log_stream:
 
             # We don't have any stdio streams to log to anymore, so we log to a file.
             # We don't override the faulthandler destination because the stream we get will proxy things
diff --git a/src/python/pants/pantsd/service/scheduler_service.py b/src/python/pants/pantsd/service/scheduler_service.py
index 767608590207..4eef2bf76488 100644
--- a/src/python/pants/pantsd/service/scheduler_service.py
+++ b/src/python/pants/pantsd/service/scheduler_service.py
@@ -4,15 +4,10 @@
 import logging
 from typing import List, Optional, Tuple, cast
 
-from pants.base.exiter import PANTS_SUCCEEDED_EXIT_CODE
-from pants.base.specs import Specs
 from pants.engine.fs import PathGlobs, Snapshot
-from pants.engine.internals.scheduler import ExecutionError, ExecutionTimeoutError
+from pants.engine.internals.scheduler import ExecutionTimeoutError
 from pants.engine.unions import UnionMembership
-from pants.goal.run_tracker import RunTracker
-from pants.init.engine_initializer import LegacyGraphScheduler, LegacyGraphSession
-from pants.option.options import Options
-from pants.option.options_bootstrapper import OptionsBootstrapper
+from pants.init.engine_initializer import LegacyGraphScheduler
 from pants.pantsd.service.fs_event_service import FSEventService
 from pants.pantsd.service.pants_service import PantsService
 
@@ -142,7 +137,7 @@ def _check_invalidation_watcher_liveness(self):
             self._logger.critical(f"The scheduler was invalidated: {e}")
             self.terminate()
 
-    def prepare_graph(self, options: Options) -> LegacyGraphSession:
+    def prepare(self) -> LegacyGraphScheduler:
         # If any nodes exist in the product graph, wait for the initial watchman event to avoid
         # racing watchman startup vs invalidation events.
         if self._fs_event_service is not None and self._scheduler.graph_len() > 0:
@@ -150,79 +145,7 @@ def prepare_graph(self, options: Options) -> LegacyGraphSession:
                 f"fs event service is running and graph_len > 0: waiting for initial watchman event"
             )
             self._fs_event_service.await_started()
-
-        global_options = options.for_global_scope()
-        build_id = RunTracker.global_instance().run_id
-        v2_ui = global_options.get("v2_ui", False)
-        use_colors = global_options.get("colors", True)
-        zipkin_trace_v2 = options.for_scope("reporting").zipkin_trace_v2
-        return self._graph_helper.new_session(
-            zipkin_trace_v2, build_id, v2_ui=v2_ui, use_colors=use_colors
-        )
-
-    def graph_run_v2(
-        self,
-        session: LegacyGraphSession,
-        specs: Specs,
-        options: Options,
-        options_bootstrapper: OptionsBootstrapper,
-    ) -> int:
-        """Perform an entire v2 run.
-
-        The exit_code in the return indicates whether any issue was encountered.
-        """
-
-        global_options = options.for_global_scope()
-        perform_loop = global_options.get("loop", False)
-        v2 = global_options.v2
-
-        if not perform_loop:
-            return self._body(session, options, options_bootstrapper, specs, v2, poll=False)
-
-        iterations = global_options.loop_max
-        exit_code = PANTS_SUCCEEDED_EXIT_CODE
-
-        while iterations and not self._state.is_terminating:
-            # NB: We generate a new "run id" per iteration of the loop in order to allow us to
-            # observe fresh values for Goals. See notes in `scheduler.rs`.
-            session.scheduler_session.new_run_id()
-            try:
-                exit_code = self._body(session, options, options_bootstrapper, specs, v2, poll=True)
-            except ExecutionError as e:
-                self._logger.warning(e)
-            iterations -= 1
-
-        return exit_code
-
-    def _body(
-        self,
-        session: LegacyGraphSession,
-        options: Options,
-        options_bootstrapper: OptionsBootstrapper,
-        specs: Specs,
-        v2: bool,
-        poll: bool,
-    ) -> int:
-        exit_code = PANTS_SUCCEEDED_EXIT_CODE
-
-        _, ambiguous_goals, v2_goals = options.goals_by_version
-
-        if v2_goals or (ambiguous_goals and v2):
-            goals = v2_goals + (ambiguous_goals if v2 else tuple())
-
-            # When polling we use a delay (only applied in cases where we have waited for something
-            # to do) in order to avoid re-running too quickly when changes arrive in clusters.
-            exit_code = session.run_goal_rules(
-                options_bootstrapper=options_bootstrapper,
-                union_membership=self._union_membership,
-                options=options,
-                goals=goals,
-                specs=specs,
-                poll=poll,
-                poll_delay=(0.1 if poll else None),
-            )
-
-        return exit_code
+        return self._graph_helper
 
     def run(self):
         """Main service entrypoint."""
diff --git a/src/rust/engine/logging/src/logger.rs b/src/rust/engine/logging/src/logger.rs
index 6f2e1a21c7dc..a831c8041019 100644
--- a/src/rust/engine/logging/src/logger.rs
+++ b/src/rust/engine/logging/src/logger.rs
@@ -16,7 +16,7 @@ use std::sync::Arc;
 
 use chrono;
 use lazy_static::lazy_static;
-use log::{log, set_logger, set_max_level, LevelFilter, Log, Metadata, Record};
+use log::{debug, log, set_logger, set_max_level, LevelFilter, Log, Metadata, Record};
 use parking_lot::Mutex;
 use simplelog::{ConfigBuilder, LevelPadding, WriteLogger};
 use tokio::task_local;
@@ -55,7 +55,9 @@ impl Logger {
         LOGGER
           .show_rust_3rdparty_logs
           .store(show_rust_3rdparty_logs, Ordering::SeqCst);
-        set_logger(&*LOGGER).expect("Error setting up global logger.");
+        if set_logger(&*LOGGER).is_err() {
+          debug!("Logging already initialized.");
+        }
       }
       Err(err) => panic!("Unrecognised log level from python: {}: {}", max_level, err),
     };
diff --git a/testprojects/pants-plugins/src/python/test_pants_plugin/tasks/lifecycle_stub_task.py b/testprojects/pants-plugins/src/python/test_pants_plugin/tasks/lifecycle_stub_task.py
index c3a8209cf82b..79ee986763ac 100644
--- a/testprojects/pants-plugins/src/python/test_pants_plugin/tasks/lifecycle_stub_task.py
+++ b/testprojects/pants-plugins/src/python/test_pants_plugin/tasks/lifecycle_stub_task.py
@@ -6,21 +6,21 @@
 from test_pants_plugin.subsystems.lifecycle_stubs import LifecycleStubs
 
 from pants.base.exception_sink import ExceptionSink
-from pants.base.exiter import Exiter
+from pants.base.exiter import ExitCode
 from pants.task.task import Task
 from pants.util.memo import memoized_property
 
 
-class MessagingExiter(Exiter):
-    """An Exiter that prints a provided message to stderr."""
+class MessagingExiter:
+    """A class that prints a provided message to stderr before dying."""
 
     def __init__(self, message):
         super().__init__()
         self._message = message
 
-    def exit(self, *args, **kwargs):
+    def __call__(self, exit_code: ExitCode):
         print(self._message, file=sys.stderr)
-        super().exit(*args, **kwargs)
+        sys.exit(exit_code)
 
 
 class LifecycleStubTask(Task):
@@ -38,8 +38,7 @@ def _lifecycle_stubs(self):
     def execute(self):
         exit_msg = self._lifecycle_stubs.add_exiter_message
         if exit_msg:
-            new_exiter = MessagingExiter(exit_msg)
-            ExceptionSink._reset_exiter(new_exiter)
+            ExceptionSink.reset_exiter(MessagingExiter(exit_msg))
 
         output_file = self._lifecycle_stubs.new_interactive_stream_output_file
         if output_file: