Perf: lazy init the bootstrap scheduler

src/python/pants/backend/BUILD

@@ -58,12 +58,14 @@ __dependents_rules__(
             "[/python/util_rules/pex_requirements.py]",
         ),
         "src/python/pants/init/plugin_resolver.py",
+        "src/python/pants/init/plugin_resolver_rules.py",
         DEFAULT_DEPENDENTS_RULES,
     ),
     (
         "[/python/util_rules/pex.py]",
         "src/python/pants/core/goals/update_build_files.py",
         "src/python/pants/init/plugin_resolver.py",
+        "src/python/pants/init/plugin_resolver_rules.py",
         DEFAULT_DEPENDENTS_RULES,
     ),
     (

src/python/pants/init/options_initializer.py

@@ -23,7 +23,6 @@
     load_build_configuration_from_source,
 )
 from pants.init.plugin_resolver import PluginResolver
-from pants.init.plugin_resolver import rules as plugin_resolver_rules
 from pants.option.bootstrap_options import DynamicRemoteOptions
 from pants.option.errors import UnknownFlagsError
 from pants.option.options import Options
@@ -106,6 +105,11 @@ def _collect_backends_requirements(backends: list[str]) -> list[str]:
 def create_bootstrap_scheduler(
     options_bootstrapper: OptionsBootstrapper, executor: PyExecutor
 ) -> BootstrapScheduler:
+    # Imported here rather than at module scope: this pulls in the Python/PEX rule modules, which
+    # are only needed to pex-resolve distribution plugins. Keeping the import lazy leaves them off
+    # the startup critical path when there is nothing to resolve.
+    from pants.init.plugin_resolver_rules import rules as plugin_resolver_rules
+
     bc_builder = BuildConfiguration.Builder()
     # To load plugins, we only need access to the Python/PEX rules.
     load_build_configuration_from_source(bc_builder, ["pants.backend.python"])
@@ -142,8 +146,9 @@ def __init__(
         options_bootstrapper: OptionsBootstrapper,
         executor: PyExecutor,
     ) -> None:
-        self._bootstrap_scheduler = create_bootstrap_scheduler(options_bootstrapper, executor)
-        self._plugin_resolver = PluginResolver(self._bootstrap_scheduler)
+        self._plugin_resolver = PluginResolver(
+            lambda: create_bootstrap_scheduler(options_bootstrapper, executor)
+        )
 
     def build_config(
         self,

src/python/pants/init/options_initializer_test.py

@@ -1,40 +1,68 @@
 # Copyright 2015 Pants project contributors (see CONTRIBUTORS.md).
 # Licensed under the Apache License, Version 2.0 (see LICENSE).
 
-import unittest
+import pytest
 
+from pants.base.exceptions import BuildConfigurationError
 from pants.engine.env_vars import CompleteEnvironmentVars
 from pants.engine.internals.scheduler import ExecutionError
+from pants.engine.unions import UnionMembership
 from pants.init.options_initializer import OptionsInitializer
+from pants.option.errors import OptionsError
 from pants.option.options_bootstrapper import OptionsBootstrapper
 from pants.testutil import rule_runner
 
 
-class OptionsInitializerTest(unittest.TestCase):
-    def test_invalid_version(self) -> None:
-        options_bootstrapper = OptionsBootstrapper.create(
-            args=["--backend-packages=[]", "--pants-version=99.99.9999"],
-            env={},
-            allow_pantsrc=False,
-        )
-
-        env = CompleteEnvironmentVars({})
-        initializer = OptionsInitializer(options_bootstrapper, rule_runner.EXECUTOR)
-        with self.assertRaises(ExecutionError):
-            initializer.build_config(options_bootstrapper, env)
-
-    def test_global_options_validation(self) -> None:
-        # Specify an invalid combination of options.
-        ob = OptionsBootstrapper.create(
-            args=["--backend-packages=[]", "--no-watch-filesystem", "--loop"],
-            env={},
-            allow_pantsrc=False,
-        )
-        env = CompleteEnvironmentVars({})
-        initializer = OptionsInitializer(ob, rule_runner.EXECUTOR)
-        with self.assertRaises(ExecutionError) as exc:
-            initializer.build_config(ob, env)
-        self.assertIn(
+def _initializer(
+    *args: str,
+) -> tuple[OptionsBootstrapper, CompleteEnvironmentVars, OptionsInitializer]:
+    ob = OptionsBootstrapper.create(
+        args=["--backend-packages=[]", *args], env={}, allow_pantsrc=False
+    )
+    env = CompleteEnvironmentVars({})
+    return ob, env, OptionsInitializer(ob, rule_runner.EXECUTOR)
+
+
+@pytest.mark.parametrize(
+    "invalid_args, message",
+    [
+        (["--pants-version=99.99.9999"], "Version mismatch"),
+        (
+            ["--no-watch-filesystem", "--loop"],
+            "The `--no-watch-filesystem` option may not be set if `--pantsd` or `--loop` is set.",
+        ),
+    ],
+)
+def test_build_config_validates_options_when_resolving_plugins(
+    invalid_args: list[str], message: str
+) -> None:
+    # With plugins configured, `build_config` resolves them through the bootstrap scheduler, which
+    # validates options as a side effect and reports failures as an `ExecutionError`.
+    ob, env, initializer = _initializer("--plugins=fake-plugin", *invalid_args)
+    with pytest.raises(ExecutionError) as exc:
+        initializer.build_config(ob, env)
+    assert message in str(exc.value)
+
+
+@pytest.mark.parametrize(
+    "invalid_args, native_exception, message",
+    [
+        (["--pants-version=99.99.9999"], BuildConfigurationError, "Version mismatch"),
+        (
+            ["--no-watch-filesystem", "--loop"],
+            OptionsError,
             "The `--no-watch-filesystem` option may not be set if `--pantsd` or `--loop` is set.",
-            str(exc.exception),
-        )
+        ),
+    ],
+)
+def test_options_validates_when_no_plugins_to_resolve(
+    invalid_args: list[str], native_exception: type[Exception], message: str
+) -> None:
+    # With nothing to resolve, plugin resolution short-circuits and `build_config` does no
+    # validation; the failure surfaces natively when full options are parsed.
+    ob, env, initializer = _initializer(*invalid_args)
+    build_config = initializer.build_config(ob, env)
+    union_membership = UnionMembership.from_rules(build_config.union_rules)
+    with pytest.raises(native_exception) as exc:
+        initializer.options(ob, env, build_config, union_membership, raise_=True)
+    assert message in str(exc.value)

src/python/pants/init/plugin_resolver.py

@@ -7,29 +7,23 @@
 import logging
 import site
 import sys
-from collections.abc import Iterable
+from collections.abc import Callable, Iterable
 from dataclasses import dataclass
-from typing import cast
+from typing import TYPE_CHECKING, cast
 
 from packaging.requirements import Requirement
 
-from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints
-from pants.backend.python.util_rules.pex import PexRequest, VenvPexProcess, create_venv_pex
-from pants.backend.python.util_rules.pex_environment import PythonExecutable
-from pants.backend.python.util_rules.pex_requirements import PexRequirements
 from pants.core.environments.rules import determine_bootstrap_environment
 from pants.engine.collection import DeduplicatedCollection
 from pants.engine.env_vars import CompleteEnvironmentVars
-from pants.engine.environment import EnvironmentName
 from pants.engine.internals.selectors import Params
 from pants.engine.internals.session import SessionValues
-from pants.engine.process import ProcessCacheScope, execute_process_or_raise
-from pants.engine.rules import QueryRule, collect_rules, implicitly, rule
 from pants.init.bootstrap_scheduler import BootstrapScheduler
 from pants.init.import_util import find_matching_distributions
-from pants.option.global_options import GlobalOptions
 from pants.option.options_bootstrapper import OptionsBootstrapper
-from pants.util.logging import LogLevel
+
+if TYPE_CHECKING:
+    from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints
 
 logger = logging.getLogger(__name__)
 
@@ -50,69 +44,6 @@ class ResolvedPluginDistributions(DeduplicatedCollection[str]):
     sort_input = True
 
 
-@rule
-async def resolve_plugins(
-    request: PluginsRequest,
-    global_options: GlobalOptions,
-) -> ResolvedPluginDistributions:
-    """This rule resolves plugins using a VenvPex, and exposes the absolute paths of their dists.
-
-    NB: This relies on the fact that PEX constructs venvs in a stable location (within the
-    `named_caches` directory), but consequently needs to disable the process cache: see the
-    ProcessCacheScope reference in the body.
-    """
-    req_strings = sorted(global_options.plugins + request.requirements)
-
-    requirements = PexRequirements(
-        req_strings_or_addrs=req_strings,
-        constraints_strings=(str(constraint) for constraint in request.constraints),
-        description_of_origin="configured Pants plugins",
-    )
-    if not requirements:
-        return ResolvedPluginDistributions()
-
-    python: PythonExecutable | None = None
-    if not request.interpreter_constraints:
-        python = PythonExecutable.fingerprinted(
-            sys.executable, ".".join(map(str, sys.version_info[:3])).encode("utf8")
-        )
-
-    plugins_pex = await create_venv_pex(
-        **implicitly(
-            PexRequest(
-                output_filename="pants_plugins.pex",
-                internal_only=True,
-                python=python,
-                requirements=requirements,
-                interpreter_constraints=request.interpreter_constraints or InterpreterConstraints(),
-                additional_args=("--preserve-pip-download-log", "pex-pip-download.log"),
-                description=f"Resolving plugins: {', '.join(req_strings)}",
-            )
-        )
-    )
-
-    # NB: We run this Process per-restart because it (intentionally) leaks named cache
-    # paths in a way that invalidates the Process-cache. See the method doc.
-    cache_scope = (
-        ProcessCacheScope.PER_SESSION
-        if global_options.plugins_force_resolve
-        else ProcessCacheScope.PER_RESTART_SUCCESSFUL
-    )
-
-    plugins_process_result = await execute_process_or_raise(
-        **implicitly(
-            VenvPexProcess(
-                plugins_pex,
-                argv=("-c", "import os, site; print(os.linesep.join(site.getsitepackages()))"),
-                description="Extracting plugin locations",
-                level=LogLevel.DEBUG,
-                cache_scope=cache_scope,
-            )
-        )
-    )
-    return ResolvedPluginDistributions(plugins_process_result.stdout.decode().strip().split("\n"))
-
-
 class PluginResolver:
     """Encapsulates the state of plugin loading.
 
@@ -122,21 +53,33 @@ class PluginResolver:
 
     def __init__(
         self,
-        scheduler: BootstrapScheduler,
+        scheduler: Callable[[], BootstrapScheduler],
         interpreter_constraints: InterpreterConstraints | None = None,
         inherit_existing_constraints: bool = True,
     ) -> None:
-        self._scheduler = scheduler
+        self._scheduler_factory = scheduler
+        self._scheduler_instance: BootstrapScheduler | None = None
         self._interpreter_constraints = interpreter_constraints
         self._inherit_existing_constraints = inherit_existing_constraints
 
+    @property
+    def _scheduler(self) -> BootstrapScheduler:
+        if self._scheduler_instance is None:
+            self._scheduler_instance = self._scheduler_factory()
+        return self._scheduler_instance
+
     def resolve(
         self,
         options_bootstrapper: OptionsBootstrapper,
         env: CompleteEnvironmentVars,
         requirements: Iterable[str] = (),
     ) -> list[str]:
         """Resolves any configured plugins and adds them to the sys.path as a side effect."""
+        requirements = tuple(requirements)
+        # Short-circuit when there is no work to be done.
+        configured_plugins = options_bootstrapper.bootstrap_options.for_global_scope().plugins
+        if not configured_plugins and not requirements:
+            return []
 
         def to_requirement(d):
             return f"{d.name}=={d.version}"
@@ -181,10 +124,3 @@ def _resolve_plugins(
             ResolvedPluginDistributions,
             session.product_request(ResolvedPluginDistributions, params)[0],
         )
-
-
-def rules():
-    return [
-        QueryRule(ResolvedPluginDistributions, [PluginsRequest, EnvironmentName]),
-        *collect_rules(),
-    ]

src/python/pants/init/plugin_resolver_rules.py

@@ -0,0 +1,87 @@
+# Copyright 2026 Pants project contributors (see CONTRIBUTORS.md).
+# Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+from __future__ import annotations
+
+import sys
+
+from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints
+from pants.backend.python.util_rules.pex import PexRequest, VenvPexProcess, create_venv_pex
+from pants.backend.python.util_rules.pex_environment import PythonExecutable
+from pants.backend.python.util_rules.pex_requirements import PexRequirements
+from pants.engine.environment import EnvironmentName
+from pants.engine.process import ProcessCacheScope, execute_process_or_raise
+from pants.engine.rules import QueryRule, collect_rules, implicitly, rule
+from pants.init.plugin_resolver import PluginsRequest, ResolvedPluginDistributions
+from pants.option.global_options import GlobalOptions
+from pants.util.logging import LogLevel
+
+
+@rule
+async def resolve_plugins(
+    request: PluginsRequest,
+    global_options: GlobalOptions,
+) -> ResolvedPluginDistributions:
+    """This rule resolves plugins using a VenvPex, and exposes the absolute paths of their dists.
+
+    NB: This relies on the fact that PEX constructs venvs in a stable location (within the
+    `named_caches` directory), but consequently needs to disable the process cache: see the
+    ProcessCacheScope reference in the body.
+    """
+    req_strings = sorted(global_options.plugins + request.requirements)
+
+    requirements = PexRequirements(
+        req_strings_or_addrs=req_strings,
+        constraints_strings=(str(constraint) for constraint in request.constraints),
+        description_of_origin="configured Pants plugins",
+    )
+    if not requirements:
+        return ResolvedPluginDistributions()
+
+    python: PythonExecutable | None = None
+    if not request.interpreter_constraints:
+        python = PythonExecutable.fingerprinted(
+            sys.executable, ".".join(map(str, sys.version_info[:3])).encode("utf8")
+        )
+
+    plugins_pex = await create_venv_pex(
+        **implicitly(
+            PexRequest(
+                output_filename="pants_plugins.pex",
+                internal_only=True,
+                python=python,
+                requirements=requirements,
+                interpreter_constraints=request.interpreter_constraints or InterpreterConstraints(),
+                additional_args=("--preserve-pip-download-log", "pex-pip-download.log"),
+                description=f"Resolving plugins: {', '.join(req_strings)}",
+            )
+        )
+    )
+
+    # NB: We run this Process per-restart because it (intentionally) leaks named cache
+    # paths in a way that invalidates the Process-cache. See the method doc.
+    cache_scope = (
+        ProcessCacheScope.PER_SESSION
+        if global_options.plugins_force_resolve
+        else ProcessCacheScope.PER_RESTART_SUCCESSFUL
+    )
+
+    plugins_process_result = await execute_process_or_raise(
+        **implicitly(
+            VenvPexProcess(
+                plugins_pex,
+                argv=("-c", "import os, site; print(os.linesep.join(site.getsitepackages()))"),
+                description="Extracting plugin locations",
+                level=LogLevel.DEBUG,
+                cache_scope=cache_scope,
+            )
+        )
+    )
+    return ResolvedPluginDistributions(plugins_process_result.stdout.decode().strip().split("\n"))
+
+
+def rules():
+    return [
+        QueryRule(ResolvedPluginDistributions, [PluginsRequest, EnvironmentName]),
+        *collect_rules(),
+    ]

src/python/pants/init/plugin_resolver_test.py

@@ -284,7 +284,7 @@ def _create_artifact(
             sys.path.insert(0, str(site_packages_path))
 
         plugin_resolver = PluginResolver(
-            bootstrap_scheduler,
+            lambda: bootstrap_scheduler,
             interpreter_constraints,
             inherit_existing_constraints=inherit_existing_constraints,
         )