Rework Goal.Options so that V2 goals work with MyPy (pantsbuild#8742)

### Problem

MyPy does not understand `Goal.Options`, e.g. `List.Options` and `Fmt.Options`, because the implementation relies on runtime creation of the class. MyPy only can understand classes declared at compile time. This is the same reason we needed to rework `pants.util.objects.enum`, `pants.util.objects.datatype`, and `pants.engine.objects.Collection`.

At the same time, we want to preserve as much of the original API as possible because it provided a useful coupling between the `Goal` class and its underlying subsystem. We need to preserve features like the goal's docstring populating `./pants help`, for example.

One of the challenges is that the original implementation requires the Subsystem inner class to have knowledge of the Goal outer class. This is not a well-supported idiom in Python: https://stackoverflow.com/questions/2024566/access-outer-class-from-inner-class-in-python.

### Attempted solutions

#### Attempt #1: pass outer Goal directly to inner Options

I started by trying to implement https://stackoverflow.com/a/7034206. The issue is that we make heavy use of `@classmethod` and `@classproperty`, so we need to be able to access `outer_cls` at the `cls` scope and this pattern only allows using it at the instance scope (`self`).

#### Attempt #2: `Goal.Options` directly instantiated with reference to `Goal`

I then tried this pattern:

```python
class List(Goal):
  name = "list"
  
  @classmethod
  def register_options(cls, register):
    ...

class ListOptions(Options):
  goal_cls = List

@rule 
def list(..., options: ListOptions) -> List:
   ...
```

This required rewriting `rules.py` to ensure that the `Options` were properly registered with the rule graph:

https://github.com/pantsbuild/pants/blob/b88dc0f7d9c0ce322714b74d78f312b4c4b847a2/src/python/pants/engine/rules.py#L138-L155

This worked fine when a rule explicitly requested the `Options` in its rule params, but it meant that rules without explicitly registered options, like `fmt` and `lint`, were no longer registered with the engine! There was no way to feed in the sibling `FmtOptions` or `LintOptions` because these would have to be explicitly created and somehow passed to the engine, even though the corresponding rules had no need to directly consume those subsystems.

### Solution: new `GoalSubsystem` class

Setting up a `Goal` will now involve two steps:
1. Register a `GoalSubsystem` that has the name of the goal, the docstring, and any options (if any).
1. Register a `Goal`, which only needs to point `subsystem_cls` to the sibling `GoalSubsystem`.

For example:

```python
class ListOptions(GoalSubsystem):
  """List all the targets."""
  name = "list"

  @classmethod
  def register_options(cls, register):
    ...

class List(Goal):
  subsystem_cls = ListOptions
```

Rules may then explicitly request the `GoalSubsystem`, e.g. `ListOptions`.

Consumers of the `Goal`, like `rules.py` and `engine_initializer.py`, simply access `Goal.subsystem_cls` to get the attributes they previously accessed directly.

#### Conceptual motivation

This strengthens the concept that Subsystems are the sole mechanism for consuming options in V2. The `GoalSubsystem` may be seen as the external API—containing all the options, the goal name, and the goal description seen by the outside world. Meanwhile, the `Goal` consumes that subsystem (just like the `black.fmt` rule consumes the `Black` subsystem) to produce a new type used internally by the engine.

### Result

MyPy now understands V2 goals, which unblocks us from using MyPy with V2 code and tests (`TestBase` and `PantsRunIntegration` are blocked by this issue).

pants-plugins/src/python/internal_backend/rules_for_testing/register.py

@@ -3,16 +3,19 @@
 
 from pants.engine.addressable import BuildFileAddresses
 from pants.engine.console import Console
-from pants.engine.goal import Goal
+from pants.engine.goal import Goal, GoalSubsystem
 from pants.engine.rules import console_rule
 
 
-class ListAndDieForTesting(Goal):
+class ListAndDieForTestingOptions(GoalSubsystem):
   """A fast and deadly variant of `./pants list`."""
-
   name = 'list-and-die-for-testing'
 
 
+class ListAndDieForTesting(Goal):
+  subsystem_cls = ListAndDieForTestingOptions
+
+
 @console_rule
 def fast_list_and_die_for_testing(
   console: Console, addresses: BuildFileAddresses

src/python/pants/backend/project_info/rules/dependencies.py

@@ -5,14 +5,14 @@
 
 from pants.base.specs import Specs
 from pants.engine.console import Console
-from pants.engine.goal import Goal, LineOriented
+from pants.engine.goal import Goal, GoalSubsystem, LineOriented
 from pants.engine.legacy.graph import HydratedTargets, TransitiveHydratedTargets
 from pants.engine.rules import console_rule
 from pants.engine.selectors import Get
 
 
 # TODO(#8762) Get this rule to feature parity with the dependencies task.
-class Dependencies(LineOriented, Goal):
+class DependenciesOptions(LineOriented, GoalSubsystem):
   name = 'fast-dependencies'
 
   @classmethod
@@ -26,10 +26,15 @@ def register_options(cls, register):
     )
 
 
+class Dependencies(Goal):
+  subsystem_cls = DependenciesOptions
+
 @console_rule
-async def dependencies(console: Console, specs: Specs, dependencies_options: Dependencies.Options) -> Dependencies:
+async def dependencies(
+  console: Console, specs: Specs, options: DependenciesOptions
+) -> Dependencies:
   addresses: Set[str] = set()
-  if dependencies_options.values.transitive:
+  if options.values.transitive:
     transitive_targets = await Get[TransitiveHydratedTargets](Specs, specs)
     addresses.update(hydrated_target.address.spec for hydrated_target in transitive_targets.closure)
     # transitive_targets.closure includes the initial target. To keep the behavior consistent with intransitive
@@ -44,7 +49,7 @@ async def dependencies(console: Console, specs: Specs, dependencies_options: Dep
       for dep in hydrated_target.dependencies
     )
 
-  with Dependencies.line_oriented(dependencies_options, console) as print_stdout:
+  with options.line_oriented(console) as print_stdout:
     for address in sorted(addresses):
       print_stdout(address)
 

src/python/pants/backend/project_info/rules/source_file_validator.py

@@ -9,7 +9,7 @@
 from pants.base.exiter import PANTS_FAILED_EXIT_CODE, PANTS_SUCCEEDED_EXIT_CODE
 from pants.engine.console import Console
 from pants.engine.fs import Digest, FilesContent
-from pants.engine.goal import Goal
+from pants.engine.goal import Goal, GoalSubsystem
 from pants.engine.legacy.graph import HydratedTarget, HydratedTargets
 from pants.engine.objects import Collection
 from pants.engine.rules import console_rule, optionable_rule, rule
@@ -33,7 +33,7 @@ class DetailLevel(Enum):
   all = "all"
 
 
-class Validate(Goal):
+class ValidateOptions(GoalSubsystem):
   name = 'validate'
 
   @classmethod
@@ -43,6 +43,10 @@ def register_options(cls, register):
              help='How much detail to emit to the console.')
 
 
+class Validate(Goal):
+  subsystem_cls = ValidateOptions
+
+
 class SourceFileValidation(Subsystem):
   options_scope = 'sourcefile-validation'
 
@@ -215,7 +219,7 @@ def get_applicable_content_pattern_names(self, path):
 # to share goal names.
 @console_rule
 async def validate(
-  console: Console, hydrated_targets: HydratedTargets, validate_options: Validate.Options
+  console: Console, hydrated_targets: HydratedTargets, validate_options: ValidateOptions
 ) -> Validate:
   per_tgt_rmrs = await MultiGet(Get(RegexMatchResults, HydratedTarget, ht) for ht in hydrated_targets)
   regex_match_results = list(itertools.chain(*per_tgt_rmrs))

src/python/pants/engine/goal.py

@@ -1,112 +1,114 @@
 # Copyright 2019 Pants project contributors (see CONTRIBUTORS.md).
 # Licensed under the Apache License, Version 2.0 (see LICENSE).
 
-from abc import ABCMeta, abstractmethod
+from abc import abstractmethod
 from contextlib import contextmanager
 from dataclasses import dataclass
+from typing import ClassVar, Type
 
 from pants.cache.cache_setup import CacheSetup
 from pants.option.optionable import Optionable
 from pants.option.scope import ScopeInfo
 from pants.subsystem.subsystem_client_mixin import SubsystemClientMixin
-from pants.util.memo import memoized_classproperty
 from pants.util.meta import classproperty
 
 
-@dataclass(frozen=True)  # type: ignore[misc] # tracked by https://github.com/python/mypy/issues/5374, which they put as high priority.
-class Goal(metaclass=ABCMeta):
-  """The named product of a `@console_rule`.
+class GoalSubsystem(SubsystemClientMixin, Optionable):
+  """The Subsystem used by `Goal`s to register the external API, meaning the goal name, the help
+  message, and any options.
 
-  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`.
+  This class should be subclassed and given a `GoalSubsystem.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 side effects (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.
+  Rules that need to consume the GoalSubsystem's options may directly request the type:
 
-  Options values for a Goal can be retrived by declaring a dependency on the relevant `Goal.Options`
-  class.
+  ```
+  @rule
+  def list(console: Console, options: ListOptions) -> List:
+    transitive = options.values.transitive
+    documented = options.values.documented
+    ...
+  ```
   """
-  exit_code: int
 
   @classproperty
   @abstractmethod
   def name(cls):
-    """The name used to select this Goal on the commandline, and for its options."""
+    """The name used to select the corresponding Goal on the commandline and the options_scope for
+    its options."""
 
   @classproperty
   def deprecated_cache_setup_removal_version(cls):
     """Optionally defines a deprecation version for a CacheSetup dependency.
 
-    If this Goal should have an associated deprecated instance of `CacheSetup` (which was implicitly
-    required by all v1 Tasks), subclasses may set this to a valid deprecation version to create
-    that association.
+    If this GoalSubsystem should have an associated deprecated instance of `CacheSetup` (which was
+    implicitly required by all v1 Tasks), subclasses may set this to a valid deprecation version to
+    create that association.
     """
     return None
 
+  @classproperty
+  def options_scope(cls):
+    return cls.name
+
   @classmethod
-  def register_options(cls, register):
-    """Register options for the `Goal.Options` of this `Goal`.
+  def subsystem_dependencies(cls):
+    # NB: `GoalSubsystem` 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 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.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().__init__()
+    self._scope = scope
+    self._scoped_options = scoped_options
+
+  @property
+  def values(self):
+    """Returns the option values."""
+    return self._scoped_options
+
+
+@dataclass(frozen=True)
+class Goal:
+  """The named product of a `@console_rule`.
 
-    Subclasses may override and call register(*args, **kwargs). Callers can retrieve the resulting
-    options values by declaring a dependency on the `Goal.Options` class.
-    """
+  This class should be subclassed and linked to a corresponding `GoalSubsystem`:
 
-  @memoized_classproperty
-  def Options(cls):
-    # NB: The naming of this property is terribly evil. But this construction allows the inner class
-    # to get a reference to the outer class, which avoids implementers needing to subclass the inner
-    # class in order to define their options values, while still allowing for the useful namespacing
-    # of `Goal.Options`.
-    outer_cls = cls
-    class _Options(SubsystemClientMixin, Optionable, _GoalOptions):
-      @classproperty
-      def options_scope(cls):
-        return outer_cls.name
-
-      @classmethod
-      def register_options(cls, register):
-        super(_Options, cls).register_options(register)
-        # Delegate to the outer class.
-        outer_cls.register_options(register)
-
-      @classmethod
-      def subsystem_dependencies(cls):
-        # NB: `Goal.Options` implements `SubsystemClientMixin` in order to allow v1 `Tasks` to
-        # depend on v2 Goals, and for `Goals` to declare a deprecated dependency on a `CacheSetup`
-        # instance for backwards compatibility purposes. But v2 Goals should _not_ have subsystem
-        # dependencies: instead, the @rules participating (transitively) in a Goal should directly
-        # declare their Subsystem deps.
-        if outer_cls.deprecated_cache_setup_removal_version:
-          dep = CacheSetup.scoped(
-              cls,
-              removal_version=outer_cls.deprecated_cache_setup_removal_version,
-              removal_hint='Goal `{}` uses an independent caching implementation, and ignores `{}`.'.format(
-                cls.options_scope,
-                CacheSetup.subscope(cls.options_scope),
-              )
-            )
-          return (dep,)
-        return tuple()
-
-      options_scope_category = ScopeInfo.GOAL
-
-      def __init__(self, scope, scoped_options):
-        # NB: This constructor is shaped to meet the contract of `Optionable(Factory).signature`.
-        super(_Options, self).__init__()
-        self._scope = scope
-        self._scoped_options = scoped_options
-
-      @property
-      def values(self):
-        """Returns the option values for these Goal.Options."""
-        return self._scoped_options
-    _Options.__doc__ = outer_cls.__doc__
-    return _Options
-
-
-class _GoalOptions(object):
-  """A marker trait for the anonymous inner `Goal.Options` classes for `Goal`s."""
+  ```
+  class ListOptions(GoalSubsystem):
+    '''List targets.'''
+    name = "list"
+
+  class List(Goal):
+    subsystem_cls = ListOptions
+  ```
+
+  Since `@console_rules` always run in order to produce side effects (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.
+  """
+  exit_code: int
+  subsystem_cls: ClassVar[Type[GoalSubsystem]]
+
+  @classproperty
+  def name(cls):
+    return cls.subsystem_cls.name
 
 
 class Outputting:
@@ -123,24 +125,20 @@ def register_options(cls, register):
     register('--output-file', metavar='<path>',
              help='Output to this file.  If unspecified, outputs to stdout.')
 
-  @classmethod
   @contextmanager
-  def output(cls, options, console):
+  def output(self, console):
     """Given Goal.Options and a Console, yields a function for writing data to stdout, or a file.
 
     The passed options instance will generally be the `Goal.Options` of an `Outputting` `Goal`.
     """
-    with cls.output_sink(options, console) as output_sink:
+    with self.output_sink(self, console) as output_sink:
       yield lambda msg: output_sink.write(msg)
 
-  @classmethod
   @contextmanager
-  def output_sink(cls, options, console):
-    if type(options) != cls.Options:
-      raise AssertionError('Expected Options for `{}`, got: {}'.format(cls.__name__, options))
+  def output_sink(self, console):
     stdout_file = None
-    if options.values.output_file:
-      stdout_file = open(options.values.output_file, 'w')
+    if self.values.output_file:
+      stdout_file = open(self.values.output_file, 'w')
       output_sink = stdout_file
     else:
       output_sink = console.stdout
@@ -159,13 +157,12 @@ def register_options(cls, register):
     register('--sep', default='\\n', metavar='<separator>',
              help='String to use to separate lines in line-oriented output.')
 
-  @classmethod
   @contextmanager
-  def line_oriented(cls, options, console):
+  def line_oriented(self, console):
     """Given Goal.Options and a Console, yields a function for printing lines to stdout or a file.
 
     The passed options instance will generally be the `Goal.Options` of an `Outputting` `Goal`.
     """
-    sep = options.values.sep.encode().decode('unicode_escape')
-    with cls.output_sink(options, console) as output_sink:
+    sep = self.values.sep.encode().decode('unicode_escape')
+    with self.output_sink(console) as output_sink:
       yield lambda msg: print(msg, file=output_sink, end=sep)

src/python/pants/engine/rules.py

@@ -136,7 +136,7 @@ def resolve_type(name):
       for p, s in rule_visitor.gets)
 
     # Register dependencies for @console_rule/Goal.
-    dependency_rules = (optionable_rule(return_type.Options),) if is_goal_cls else None
+    dependency_rules = (optionable_rule(return_type.subsystem_cls),) if is_goal_cls else None
 
     # Set a default name for Goal classes if one is not explicitly provided
     if is_goal_cls and name is None:

src/python/pants/fs/fs_test.py

@@ -14,7 +14,7 @@
   MaterializeDirectoryResult,
   Workspace,
 )
-from pants.engine.goal import Goal
+from pants.engine.goal import Goal, GoalSubsystem
 from pants.engine.rules import RootRule, console_rule
 from pants.engine.selectors import Get
 from pants.fs.fs import is_child_of
@@ -27,10 +27,14 @@ class MessageToConsoleRule:
   input_files_content: InputFilesContent
 
 
-class MockWorkspaceGoal(Goal):
+class MockWorkspaceGoalOptions(GoalSubsystem):
   name = 'mock-workspace-goal'
 
 
+class MockWorkspaceGoal(Goal):
+  subsystem_cls = MockWorkspaceGoalOptions
+
+
 @console_rule
 async def workspace_console_rule(console: Console, workspace: Workspace, msg: MessageToConsoleRule) -> MockWorkspaceGoal:
   digest = await Get(Digest, InputFilesContent, msg.input_files_content)