diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index a70452784..64661c675 100644
--- a/.github/workflows/ci.yaml
+++ b/.github/workflows/ci.yaml
@@ -8,7 +8,7 @@ on:
 jobs:
   pytest:
     uses: colcon/ci/.github/workflows/pytest.yaml@main
-    with:
-      codecov: true
+    secrets:
+      CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
   bootstrap:
     uses: ./.github/workflows/bootstrap.yaml
diff --git a/colcon_core/__init__.py b/colcon_core/__init__.py
index 4654486c3..4f82143e3 100644
--- a/colcon_core/__init__.py
+++ b/colcon_core/__init__.py
@@ -1,4 +1,4 @@
 # Copyright 2016-2020 Dirk Thomas
 # Licensed under the Apache License, Version 2.0
 
-__version__ = '0.15.2'
+__version__ = '0.16.1'
diff --git a/colcon_core/argument_parser/__init__.py b/colcon_core/argument_parser/__init__.py
index 44f743133..21f069489 100644
--- a/colcon_core/argument_parser/__init__.py
+++ b/colcon_core/argument_parser/__init__.py
@@ -42,7 +42,7 @@ def decorate_argument_parser(self, *, parser):
         raise NotImplementedError()
 
 
-def get_argument_parser_extensions():
+def get_argument_parser_extensions(*, group_name=None):
     """
     Get the available argument parser extensions.
 
@@ -50,7 +50,9 @@ def get_argument_parser_extensions():
 
     :rtype: OrderedDict
     """
-    extensions = instantiate_extensions(__name__)
+    if group_name is None:
+        group_name = __name__
+    extensions = instantiate_extensions(group_name)
     for name, extension in extensions.items():
         extension.ARGUMENT_PARSER_DECORATOR_NAME = name
     return order_extensions_by_priority(extensions)
diff --git a/colcon_core/command.py b/colcon_core/command.py
index 07dddaf1c..9c94118c6 100644
--- a/colcon_core/command.py
+++ b/colcon_core/command.py
@@ -52,6 +52,7 @@
 from colcon_core.argument_parser import decorate_argument_parser  # noqa: E402 E501 I100 I202
 from colcon_core.argument_parser import SuppressUsageOutput  # noqa: E402
 from colcon_core.extension_point import load_extension_points  # noqa: E402
+from colcon_core.feature_flags import check_implemented_flags  # noqa: E402
 from colcon_core.location import create_log_path  # noqa: E402
 from colcon_core.location import get_log_path  # noqa: E402
 from colcon_core.location import set_default_config_path  # noqa: E402
@@ -86,12 +87,14 @@ def register_command_exit_handler(handler):
 
     :param handler: The callable
     """
-    global _command_exit_handlers
     if handler not in _command_exit_handlers:
         _command_exit_handlers.append(handler)
 
 
-def main(*, command_name='colcon', argv=None):
+def main(
+    *, command_name='colcon', argv=None, verb_group_name=None,
+    environment_variable_group_name=None,
+):
     """
     Execute the main logic of the command.
 
@@ -113,9 +116,11 @@ def main(*, command_name='colcon', argv=None):
     :param list argv: The list of arguments
     :returns: The return code
     """
-    global _command_exit_handlers
     try:
-        return _main(command_name=command_name, argv=argv)
+        return _main(
+            command_name=command_name, argv=argv,
+            verb_group_name=verb_group_name,
+            environment_variable_group_name=environment_variable_group_name)
     except KeyboardInterrupt:
         return signal.SIGINT
     finally:
@@ -125,8 +130,9 @@ def main(*, command_name='colcon', argv=None):
             handler()
 
 
-def _main(*, command_name, argv):
-    global colcon_logger
+def _main(
+    *, command_name, argv, verb_group_name, environment_variable_group_name,
+):
     # default log level, for searchability: COLCON_LOG_LEVEL
     colcon_logger.setLevel(logging.WARNING)
     set_logger_level_from_env(
@@ -135,14 +141,17 @@ def _main(*, command_name, argv):
         'Command line arguments: {argv}'
         .format(argv=argv if argv is not None else sys.argv))
 
+    # warn about any specified feature flags that are already implemented
+    check_implemented_flags()
+
     # set default locations for config files, for searchability: COLCON_HOME
     set_default_config_path(
         path=(Path('~') / f'.{command_name}').expanduser(),
         env_var=f'{command_name}_HOME'.upper())
 
-    parser = create_parser('colcon_core.environment_variable')
+    parser = create_parser(environment_variable_group_name)
 
-    verb_extensions = get_verb_extensions()
+    verb_extensions = get_verb_extensions(group_name=verb_group_name)
 
     # add subparsers for all verb extensions but without arguments for now
     subparser = create_subparser(
@@ -206,7 +215,7 @@ def _main(*, command_name, argv):
     return verb_main(context, colcon_logger)
 
 
-def create_parser(environment_variables_group_name):
+def create_parser(environment_variables_group_name=None):
     """
     Create the argument parser.
 
@@ -226,20 +235,30 @@ class CustomArgumentParser(argparse.ArgumentParser):
 
         def _parse_optional(self, arg_string):
             result = super()._parse_optional(arg_string)
-            if result == (None, arg_string, None):
+            # Up until https://github.com/python/cpython/pull/114180 ,
+            # _parse_optional() returned a 3-tuple when it couldn't classify
+            # the option.  As of that PR (which is in Python 3.13, and
+            # backported to Python 3.12), it returns a 4-tuple.  Check for
+            # either here.
+            if result in (
+                (None, arg_string, None),
+                (None, arg_string, None, None),
+            ):
                 # in the case there the arg is classified as an unknown 'O'
                 # override that and classify it as an 'A'
                 return None
             return result
 
+    epilog = get_environment_variables_epilog(environment_variables_group_name)
+    if epilog:
+        epilog += '\n\n'
+    epilog += READTHEDOCS_MESSAGE
+
     # top level parser
     parser = CustomArgumentParser(
         prog=get_prog_name(),
         formatter_class=CustomFormatter,
-        epilog=(
-            get_environment_variables_epilog(
-                environment_variables_group_name
-            ) + '\n\n' + READTHEDOCS_MESSAGE))
+        epilog=epilog)
 
     # enable introspecting and intercepting all command line arguments
     parser = decorate_argument_parser(parser)
@@ -256,9 +275,28 @@ def get_prog_name():
     if basename == '__main__.py':
         # use the module name in case the script was invoked with python -m ...
         prog = os.path.basename(os.path.dirname(prog))
-    elif shutil.which(basename) == prog:
-        # use basename only if it is on the PATH
-        prog = basename
+    else:
+        default_prog = shutil.which(basename) or ''
+        default_ext = os.path.splitext(default_prog)[1]
+        real_prog = prog
+        if (
+            sys.platform == 'win32' and
+            os.path.splitext(real_prog)[1] != default_ext
+        ):
+            # On Windows, setuptools entry points drop the file extension from
+            # argv[0], but shutil.which does not. If the two don't end in the
+            # same extension, try appending the shutil extension for a better
+            # chance at matching.
+            real_prog += default_ext
+        try:
+            # The os.path.samefile requires that both files exist on disk, but
+            # has the advantage of working around symlinks, UNC-style paths,
+            # DOS 8.3 path atoms, and path normalization.
+            if os.path.samefile(default_prog, real_prog):
+                # use basename only if it is on the PATH
+                prog = basename
+        except (FileNotFoundError, NotADirectoryError):
+            pass
     return prog
 
 
@@ -276,7 +314,7 @@ def _split_lines(self, text, width):
         return lines
 
 
-def get_environment_variables_epilog(group_name):
+def get_environment_variables_epilog(group_name=None):
     """
     Get a message enumerating the registered environment variables.
 
@@ -285,8 +323,12 @@ def get_environment_variables_epilog(group_name):
     :returns: The message for the argument parser epilog
     :rtype: str
     """
+    if group_name is None:
+        group_name = 'colcon_core.environment_variable'
     # list environment variables with descriptions
     entry_points = load_extension_points(group_name)
+    if not entry_points:
+        return ''
     env_vars = {
         env_var.name: env_var.description for env_var in entry_points.values()}
     epilog_lines = []
@@ -375,9 +417,6 @@ def create_subparser(parser, cmd_name, verb_extensions, *, attribute):
       selected verb
     :returns: The special action object
     """
-    global colcon_logger
-    assert verb_extensions, 'No verb extensions'
-
     # list of available verbs with their descriptions
     verbs = []
     for name, extension in verb_extensions.items():
@@ -387,9 +426,9 @@ def create_subparser(parser, cmd_name, verb_extensions, *, attribute):
     # add subparser with description of verb extensions
     subparser = parser.add_subparsers(
         title=f'{cmd_name} verbs',
-        description='\n'.join(verbs),
+        description='\n'.join(verbs) or None,
         dest=attribute,
-        help=f'call `{cmd_name} VERB -h` for specific help',
+        help=f'call `{cmd_name} VERB -h` for specific help' if verbs else None,
     )
     return subparser
 
diff --git a/colcon_core/environment/__init__.py b/colcon_core/environment/__init__.py
index d57adbf3b..a7324d017 100644
--- a/colcon_core/environment/__init__.py
+++ b/colcon_core/environment/__init__.py
@@ -47,7 +47,7 @@ def create_environment_hooks(self, prefix_path, pkg_name):
         raise NotImplementedError()
 
 
-def get_environment_extensions():
+def get_environment_extensions(*, group_name=None):
     """
     Get the available environment extensions.
 
@@ -55,7 +55,9 @@ def get_environment_extensions():
 
     :rtype: OrderedDict
     """
-    extensions = instantiate_extensions(__name__)
+    if group_name is None:
+        group_name = __name__
+    extensions = instantiate_extensions(group_name)
     for name in list(extensions.keys()):
         extension = extensions[name]
         extension.ENVIRONMENT_NAME = name
diff --git a/colcon_core/event_handler/__init__.py b/colcon_core/event_handler/__init__.py
index a51756c5b..437894626 100644
--- a/colcon_core/event_handler/__init__.py
+++ b/colcon_core/event_handler/__init__.py
@@ -44,7 +44,7 @@ def __call__(self, event):
         raise NotImplementedError()
 
 
-def get_event_handler_extensions(*, context):
+def get_event_handler_extensions(*, context, group_name=None):
     """
     Get the available event handler extensions.
 
@@ -52,7 +52,9 @@ def get_event_handler_extensions(*, context):
 
     :rtype: OrderedDict
     """
-    extensions = instantiate_extensions(__name__)
+    if group_name is None:
+        group_name = __name__
+    extensions = instantiate_extensions(group_name)
     for name, extension in extensions.items():
         extension.EVENT_HANDLER_NAME = name
         extension.context = context
diff --git a/colcon_core/executor/__init__.py b/colcon_core/executor/__init__.py
index aca035758..c0bdaccf9 100644
--- a/colcon_core/executor/__init__.py
+++ b/colcon_core/executor/__init__.py
@@ -197,7 +197,7 @@ def _flush(self):
         self._event_controller.flush()
 
 
-def get_executor_extensions():
+def get_executor_extensions(*, group_name=None):
     """
     Get the available executor extensions.
 
@@ -206,7 +206,9 @@ def get_executor_extensions():
 
     :rtype: OrderedDict
     """
-    extensions = instantiate_extensions(__name__)
+    if group_name is None:
+        group_name = __name__
+    extensions = instantiate_extensions(group_name)
     for name, extension in extensions.items():
         extension.EXECUTOR_NAME = name
     return order_extensions_grouped_by_priority(extensions)
diff --git a/colcon_core/executor/sequential.py b/colcon_core/executor/sequential.py
index 1aab44e79..eeea8e1e3 100644
--- a/colcon_core/executor/sequential.py
+++ b/colcon_core/executor/sequential.py
@@ -10,6 +10,7 @@
 from colcon_core.executor import ExecutorExtensionPoint
 from colcon_core.executor import OnError
 from colcon_core.logging import colcon_logger
+from colcon_core.logging import get_effective_console_level
 from colcon_core.plugin_system import satisfies_version
 from colcon_core.subprocess import new_event_loop
 from colcon_core.subprocess import SIGINT_RESULT
@@ -32,7 +33,8 @@ def __init__(self):  # noqa: D107
     def execute(self, args, jobs, *, on_error=OnError.interrupt):  # noqa: D102
         # avoid debug message from asyncio when colcon uses debug log level
         asyncio_logger = logging.getLogger('asyncio')
-        asyncio_logger.setLevel(logging.INFO)
+        log_level = get_effective_console_level(colcon_logger)
+        asyncio_logger.setLevel(log_level)
 
         rc = 0
         loop = new_event_loop()
diff --git a/colcon_core/extension_point.py b/colcon_core/extension_point.py
index c724d2213..e04ee0fdc 100644
--- a/colcon_core/extension_point.py
+++ b/colcon_core/extension_point.py
@@ -3,8 +3,11 @@
 # Licensed under the Apache License, Version 2.0
 
 from collections import defaultdict
+from itertools import chain
 import os
+import sys
 import traceback
+import warnings
 
 try:
     from importlib.metadata import distributions
@@ -19,13 +22,15 @@
 from colcon_core.environment_variable import EnvironmentVariable
 from colcon_core.logging import colcon_logger
 
-"""Environment variable to block extensions"""
-EXTENSION_BLOCKLIST_ENVIRONMENT_VARIABLE = EnvironmentVariable(
+_EXTENSION_BLOCKLIST_ENVIRONMENT_VARIABLE = EnvironmentVariable(
     'COLCON_EXTENSION_BLOCKLIST',
     'Block extensions which should not be used')
 
-logger = colcon_logger.getChild(__name__)
+"""Environment variable to block extensions"""
+EXTENSION_BLOCKLIST_ENVIRONMENT_VARIABLE = \
+    _EXTENSION_BLOCKLIST_ENVIRONMENT_VARIABLE
 
+logger = colcon_logger.getChild(__name__)
 
 """
 The group name for entry points identifying colcon extension points.
@@ -36,6 +41,8 @@
 """
 EXTENSION_POINT_GROUP_NAME = 'colcon_core.extension_point'
 
+_ENTRY_POINTS_CACHE = []
+
 
 def _get_unique_distributions():
     seen = set()
@@ -46,6 +53,50 @@ def _get_unique_distributions():
             yield dist
 
 
+def _get_entry_points():
+    for dist in _get_unique_distributions():
+        for entry_point in dist.entry_points:
+            # Modern EntryPoint instances should already have this set
+            if not hasattr(entry_point, 'dist'):
+                entry_point.dist = dist
+            yield entry_point
+
+
+def _get_cached_entry_points():
+    if not _ENTRY_POINTS_CACHE:
+        if sys.version_info >= (3, 10):
+            # We prefer using importlib.metadata.entry_points because it
+            # has an internal optimization which allows us to load the entry
+            # points without reading the individual PKG-INFO files, while
+            # still visiting each unique distribution only once.
+            all_entry_points = entry_points()
+            if isinstance(all_entry_points, dict):
+                # Prior to Python 3.12, entry_points returned a (deprecated)
+                # dict. Unfortunately, the "future-proof" recommended
+                # pattern is to add filter parameters, but we actually
+                # want to cache everything so that doesn't work here.
+                with warnings.catch_warnings():
+                    warnings.filterwarnings(
+                        'ignore',
+                        'SelectableGroups dict interface is deprecated',
+                        DeprecationWarning,
+                        module=__name__)
+                    all_entry_points = chain.from_iterable(
+                        all_entry_points.values())
+            _ENTRY_POINTS_CACHE.extend(all_entry_points)
+        else:
+            # If we don't have Python 3.10, we must read each PKG-INFO to
+            # get the name of the distribution so that we can skip the
+            # "shadowed" distributions properly.
+            _ENTRY_POINTS_CACHE.extend(_get_entry_points())
+    return _ENTRY_POINTS_CACHE
+
+
+def clear_entry_point_cache():
+    """Purge the entry point cache."""
+    _ENTRY_POINTS_CACHE.clear()
+
+
 def get_all_extension_points():
     """
     Get all extension points related to `colcon` and any of its extensions.
@@ -59,23 +110,24 @@ def get_all_extension_points():
     colcon_extension_points = get_extension_points(EXTENSION_POINT_GROUP_NAME)
     colcon_extension_points.setdefault(EXTENSION_POINT_GROUP_NAME, None)
 
-    entry_points = defaultdict(dict)
-    for dist in _get_unique_distributions():
-        for entry_point in dist.entry_points:
-            # skip groups which are not registered as extension points
-            if entry_point.group not in colcon_extension_points:
-                continue
-
-            if entry_point.name in entry_points[entry_point.group]:
-                previous = entry_points[entry_point.group][entry_point.name]
-                logger.error(
-                    f"Entry point '{entry_point.group}.{entry_point.name}' is "
-                    f"declared multiple times, '{entry_point.value}' "
-                    f"from '{dist._path}' "
-                    f"overwriting '{previous}'")
-            entry_points[entry_point.group][entry_point.name] = \
-                (entry_point.value, dist.metadata['Name'], dist.version)
-    return entry_points
+    extension_points = defaultdict(dict)
+    for entry_point in _get_cached_entry_points():
+        if entry_point.group not in colcon_extension_points:
+            continue
+
+        dist_metadata = entry_point.dist.metadata
+        ep_tuple = (
+            entry_point.value,
+            dist_metadata['Name'], dist_metadata['Version'],
+        )
+        if entry_point.name in extension_points[entry_point.group]:
+            previous = extension_points[entry_point.group][entry_point.name]
+            logger.error(
+                f"Entry point '{entry_point.group}.{entry_point.name}' is "
+                f"declared multiple times, '{ep_tuple}' "
+                f"overwriting '{previous}'")
+        extension_points[entry_point.group][entry_point.name] = ep_tuple
+    return extension_points
 
 
 def get_extension_points(group):
@@ -87,16 +139,9 @@ def get_extension_points(group):
     :rtype: dict
     """
     extension_points = {}
-    try:
-        # Python 3.10 and newer
-        query = entry_points(group=group)
-    except TypeError:
-        query = (
-            entry_point
-            for dist in _get_unique_distributions()
-            for entry_point in dist.entry_points
-            if entry_point.group == group)
-    for entry_point in query:
+    for entry_point in _get_cached_entry_points():
+        if entry_point.group != group:
+            continue
         if entry_point.name in extension_points:
             previous_entry_point = extension_points[entry_point.name]
             logger.error(
@@ -163,3 +208,16 @@ def load_extension_point(name, value, group):
                 'The entry point name is listed in the environment variable '
                 f"'{EXTENSION_BLOCKLIST_ENVIRONMENT_VARIABLE.name}'")
     return EntryPoint(name, value, group).load()
+
+
+def override_blocklist_variable(variable):
+    """
+    Override the blocklist environment variable.
+
+    :param EnvironmentVariable variable: The new blocklist environment
+      variable, or None to reset to default.
+    """
+    if variable is None:
+        variable = _EXTENSION_BLOCKLIST_ENVIRONMENT_VARIABLE
+    global EXTENSION_BLOCKLIST_ENVIRONMENT_VARIABLE
+    EXTENSION_BLOCKLIST_ENVIRONMENT_VARIABLE = variable
diff --git a/colcon_core/feature_flags.py b/colcon_core/feature_flags.py
new file mode 100644
index 000000000..56bb5bec4
--- /dev/null
+++ b/colcon_core/feature_flags.py
@@ -0,0 +1,71 @@
+# Copyright 2024 Open Source Robotics Foundation, Inc.
+# Licensed under the Apache License, Version 2.0
+
+import os
+
+from colcon_core.environment_variable import EnvironmentVariable
+from colcon_core.logging import colcon_logger
+
+logger = colcon_logger.getChild(__name__)
+
+"""Environment variable to enable feature flags"""
+FEATURE_FLAGS_ENVIRONMENT_VARIABLE = EnvironmentVariable(
+    'COLCON_FEATURE_FLAGS',
+    'Enable pre-production features and behaviors')
+
+_REPORTED_USES = set()
+
+IMPLEMENTED_FLAGS = set()
+
+
+def check_implemented_flags():
+    """Check for and warn about flags which have been implemented."""
+    implemented = IMPLEMENTED_FLAGS.intersection(get_feature_flags())
+    if implemented:
+        logger.warning(
+            'The following feature flags have been implemented and should no '
+            'longer be specified in '
+            f'{FEATURE_FLAGS_ENVIRONMENT_VARIABLE.name}: '
+            f"{','.join(implemented)}")
+
+
+def get_feature_flags():
+    """
+    Retrieve all enabled feature flags.
+
+    :returns: List of enabled flags
+    :rtype: list
+    """
+    return [
+        flag for flag in (
+            os.environ.get(FEATURE_FLAGS_ENVIRONMENT_VARIABLE.name) or ''
+        ).split(os.pathsep) if flag
+    ]
+
+
+def is_feature_flag_set(flag):
+    """
+    Determine if a specific feature flag is enabled.
+
+    Feature flags are case-sensitive and separated by the os-specific path
+    separator character.
+
+    :param str flag: Name of the flag to search for
+
+    :returns: True if the flag is set
+    :rtype: bool
+    """
+    if flag in IMPLEMENTED_FLAGS:
+        return True
+    elif flag in get_feature_flags():
+        if flag not in _REPORTED_USES:
+            if not _REPORTED_USES:
+                logger.warning(
+                    'One or more feature flags have been enabled using the '
+                    f'{FEATURE_FLAGS_ENVIRONMENT_VARIABLE.name} environment '
+                    'variable. These features may be unstable and may change '
+                    'API or behavior at any time.')
+            logger.warning(f'Enabling feature: {flag}')
+            _REPORTED_USES.add(flag)
+        return True
+    return False
diff --git a/colcon_core/logging.py b/colcon_core/logging.py
index 0a555b6f0..b50ccc945 100644
--- a/colcon_core/logging.py
+++ b/colcon_core/logging.py
@@ -95,15 +95,15 @@ def filter(self, record):  # noqa: A003
         if isinstance(handler, logging.StreamHandler):
             formatter = handler.formatter
             # filter colcon specific log messages from default stream handler
-            handler.addFilter(Filter(colcon_logger.name))
+            handler.addFilter(Filter(logger.name))
 
     # add a stream handler replacing the one filtered on the root logger
     handler = logging.StreamHandler()
     if formatter:
         # use same formatter as for stream handler
         handler.setFormatter(formatter)
-    handler.setLevel(colcon_logger.getEffectiveLevel())
-    colcon_logger.addHandler(handler)
+    handler.setLevel(logger.getEffectiveLevel())
+    logger.addHandler(handler)
 
     # add a file handler writing all log levels
     handler = logging.FileHandler(str(path))
@@ -121,9 +121,26 @@ def format_message_with_relative_time(record):
         # use same formatter as for stream handler
         handler.setFormatter(formatter)
     handler.setLevel(1)
-    colcon_logger.addHandler(handler)
+    logger.addHandler(handler)
 
     # change the logger to handle all levels
-    colcon_logger.setLevel(1)
+    logger.setLevel(1)
 
     return handler
+
+
+def get_effective_console_level(logger):
+    """
+    Determine the effective log level of to the console.
+
+    On a typical logger, this is the same as getEffectiveLevel(). After a call
+    to add_file_handler, this will continue to return the same level though
+    getEffectiveLevel() will now always return ``1``.
+
+    :param logger: The logger to inspect
+    :returns: the log level
+    """
+    for handler in logger.handlers:
+        if isinstance(handler, logging.StreamHandler):
+            return handler.level
+    return logger.getEffectiveLevel()
diff --git a/colcon_core/package_augmentation/__init__.py b/colcon_core/package_augmentation/__init__.py
index 3b579292f..e741292a1 100644
--- a/colcon_core/package_augmentation/__init__.py
+++ b/colcon_core/package_augmentation/__init__.py
@@ -65,7 +65,7 @@ def augment_package(
         raise NotImplementedError()
 
 
-def get_package_augmentation_extensions():
+def get_package_augmentation_extensions(*, group_name=None):
     """
     Get the available package augmentation extensions.
 
@@ -73,7 +73,9 @@ def get_package_augmentation_extensions():
 
     :rtype: OrderedDict
     """
-    extensions = instantiate_extensions(__name__)
+    if group_name is None:
+        group_name = __name__
+    extensions = instantiate_extensions(group_name)
     for name, extension in extensions.items():
         extension.PACKAGE_AUGMENTATION_NAME = name
     return order_extensions_by_priority(extensions)
diff --git a/colcon_core/package_decorator.py b/colcon_core/package_decorator.py
index 256b7c859..3226da1a1 100644
--- a/colcon_core/package_decorator.py
+++ b/colcon_core/package_decorator.py
@@ -46,8 +46,9 @@ def add_recursive_dependencies(
 
     :param set decorators: The known packages to consider
     :param Iterable[str] direct_categories: The names of the direct categories
-    :param Iterable[str] recursive_categories: The names of the recursive
-    categories
+    :param Iterable[str]|Mapping[str, Iterable[str]] recursive_categories:
+      The names of the recursive categories, optionally mapped from the
+      immediate upstream category which included the dependency
     """
     descriptors = [decorator.descriptor for decorator in decorators]
     for decorator in decorators:
diff --git a/colcon_core/package_descriptor.py b/colcon_core/package_descriptor.py
index 0e61006dc..bffdf9e5f 100644
--- a/colcon_core/package_descriptor.py
+++ b/colcon_core/package_descriptor.py
@@ -2,6 +2,7 @@
 # Licensed under the Apache License, Version 2.0
 
 from collections import defaultdict
+from collections.abc import Mapping
 from copy import deepcopy
 import os
 from pathlib import Path
@@ -105,12 +106,15 @@ def get_recursive_dependencies(
           consider
         :param Iterable[str] direct_categories: The names of the direct
           categories
-        :param Iterable[str] recursive_categories: The names of the recursive
-          categories
+        :param Iterable[str]|Mapping[str, Iterable[str]] recursive_categories:
+          The names of the recursive categories, optionally mapped from the
+          immediate upstream category which included the dependency
         :returns: The dependencies
         :rtype: set[DependencyDescriptor]
         :raises AssertionError: if a package lists itself as a dependency
         """
+        if not isinstance(recursive_categories, Mapping):
+            recursive_categories = defaultdict(lambda: recursive_categories)
         # the following variable only exists for faster access within the loop
         descriptors_by_name = defaultdict(set)
         for d in descriptors:
@@ -132,11 +136,17 @@ def get_recursive_dependencies(
                 descs = descriptors_by_name[dep]
                 if not descs:
                     continue
+                categories = set()
+                for category in dep.metadata['categories']:
+                    cats = recursive_categories.get(category)
+                    if cats is None:
+                        categories = None
+                        break
+                    categories.update(cats)
                 # recursing into the same function of the dependency descriptor
                 # queue recursive dependencies
                 for d in descs:
-                    queue |= d.get_dependencies(
-                        categories=recursive_categories)
+                    queue |= d.get_dependencies(categories=categories)
                 # add the depth
                 dep.metadata['depth'] = depth
                 # add dependency to result set
diff --git a/colcon_core/package_discovery/__init__.py b/colcon_core/package_discovery/__init__.py
index fc60f1470..c16f3b1d3 100644
--- a/colcon_core/package_discovery/__init__.py
+++ b/colcon_core/package_discovery/__init__.py
@@ -84,7 +84,7 @@ def discover(self, *, args, identification_extensions):
         raise NotImplementedError()
 
 
-def get_package_discovery_extensions():
+def get_package_discovery_extensions(*, group_name=None):
     """
     Get the available package discovery extensions.
 
@@ -92,7 +92,9 @@ def get_package_discovery_extensions():
 
     :rtype: OrderedDict
     """
-    extensions = instantiate_extensions(__name__)
+    if group_name is None:
+        group_name = __name__
+    extensions = instantiate_extensions(group_name)
     for name, extension in extensions.items():
         extension.PACKAGE_DISCOVERY_NAME = name
     return order_extensions_by_priority(extensions)
diff --git a/colcon_core/package_identification/__init__.py b/colcon_core/package_identification/__init__.py
index 11e17d73f..01dd8a2ce 100644
--- a/colcon_core/package_identification/__init__.py
+++ b/colcon_core/package_identification/__init__.py
@@ -64,7 +64,7 @@ def identify(self, desc: PackageDescriptor):
         raise NotImplementedError()
 
 
-def get_package_identification_extensions():
+def get_package_identification_extensions(*, group_name=None):
     """
     Get the available package identification extensions.
 
@@ -73,7 +73,9 @@ def get_package_identification_extensions():
 
     :rtype: OrderedDict
     """
-    extensions = instantiate_extensions(__name__)
+    if group_name is None:
+        group_name = __name__
+    extensions = instantiate_extensions(group_name)
     for name, extension in extensions.items():
         extension.PACKAGE_IDENTIFICATION_NAME = name
     return order_extensions_grouped_by_priority(extensions)
diff --git a/colcon_core/package_selection/__init__.py b/colcon_core/package_selection/__init__.py
index 6b1bdc9f0..e76a474b0 100644
--- a/colcon_core/package_selection/__init__.py
+++ b/colcon_core/package_selection/__init__.py
@@ -85,7 +85,7 @@ def add_arguments(parser):
     _add_package_selection_arguments(parser)
 
 
-def get_package_selection_extensions():
+def get_package_selection_extensions(*, group_name=None):
     """
     Get the available package selection extensions.
 
@@ -93,7 +93,9 @@ def get_package_selection_extensions():
 
     :rtype: OrderedDict
     """
-    extensions = instantiate_extensions(__name__)
+    if group_name is None:
+        group_name = __name__
+    extensions = instantiate_extensions(group_name)
     for name, extension in extensions.items():
         extension.PACKAGE_SELECTION_NAME = name
     return order_extensions_by_priority(extensions)
@@ -136,8 +138,9 @@ def get_packages(
     :param additional_argument_names: A list of additional arguments to
       consider
     :param Iterable[str] direct_categories: The names of the direct categories
-    :param Iterable[str] recursive_categories: The names of the recursive
-      categories
+    :param Iterable[str]|Mapping[str, Iterable[str]] recursive_categories:
+      The names of the recursive categories, optionally mapped from the
+      immediate upstream category which included the dependency
     :rtype: list
     :raises RuntimeError: if the returned set of packages contains duplicates
       package names
diff --git a/colcon_core/prefix_path/__init__.py b/colcon_core/prefix_path/__init__.py
index 15aae0b2e..4fa34eb88 100644
--- a/colcon_core/prefix_path/__init__.py
+++ b/colcon_core/prefix_path/__init__.py
@@ -40,7 +40,7 @@ def extend_prefix_path(self, paths):
         raise NotImplementedError()
 
 
-def get_prefix_path_extensions():
+def get_prefix_path_extensions(*, group_name=None):
     """
     Get the available prefix path extensions.
 
@@ -49,7 +49,9 @@ def get_prefix_path_extensions():
 
     :rtype: OrderedDict
     """
-    extensions = instantiate_extensions(__name__)
+    if group_name is None:
+        group_name = __name__
+    extensions = instantiate_extensions(group_name)
     for name, extension in extensions.items():
         extension.PREFIX_PATH_NAME = name
     return order_extensions_grouped_by_priority(extensions)
diff --git a/colcon_core/shell/__init__.py b/colcon_core/shell/__init__.py
index faa103ff7..7924b7953 100644
--- a/colcon_core/shell/__init__.py
+++ b/colcon_core/shell/__init__.py
@@ -273,7 +273,7 @@ async def generate_command_environment(
         raise NotImplementedError()
 
 
-def get_shell_extensions():
+def get_shell_extensions(*, group_name=None):
     """
     Get the available shell extensions.
 
@@ -282,7 +282,9 @@ def get_shell_extensions():
 
     :rtype: OrderedDict
     """
-    extensions = instantiate_extensions(__name__)
+    if group_name is None:
+        group_name = __name__
+    extensions = instantiate_extensions(group_name)
     for name, extension in extensions.items():
         extension.SHELL_NAME = name
     return order_extensions_grouped_by_priority(extensions)
@@ -593,7 +595,7 @@ def find_installed_packages(self, install_base: Path):
         raise NotImplementedError()
 
 
-def get_find_installed_packages_extensions():
+def get_find_installed_packages_extensions(*, group_name=None):
     """
     Get the available package identification extensions.
 
@@ -602,7 +604,10 @@ def get_find_installed_packages_extensions():
 
     :rtype: OrderedDict
     """
-    extensions = instantiate_extensions(__name__ + '.find_installed_packages')
+    if group_name is None:
+        group_name = __name__
+    extensions = instantiate_extensions(
+        group_name + '.find_installed_packages')
     for name, extension in extensions.items():
         extension.PACKAGE_IDENTIFICATION_NAME = name
     return order_extensions_grouped_by_priority(extensions)
diff --git a/colcon_core/shell/template/prefix_util.py.em b/colcon_core/shell/template/prefix_util.py.em
index 6f96913c3..9ed25d2fd 100644
--- a/colcon_core/shell/template/prefix_util.py.em
+++ b/colcon_core/shell/template/prefix_util.py.em
@@ -15,9 +15,9 @@ FORMAT_STR_SET_ENV_VAR = '@(shell_extension.FORMAT_STR_SET_ENV_VAR)'
 @{assert shell_extension.FORMAT_STR_USE_ENV_VAR is not None}@
 FORMAT_STR_USE_ENV_VAR = '@(shell_extension.FORMAT_STR_USE_ENV_VAR)'
 @{assert shell_extension.FORMAT_STR_INVOKE_SCRIPT is not None}@
-FORMAT_STR_INVOKE_SCRIPT = '@(shell_extension.FORMAT_STR_INVOKE_SCRIPT)'
-FORMAT_STR_REMOVE_LEADING_SEPARATOR = '@(shell_extension.FORMAT_STR_REMOVE_LEADING_SEPARATOR)'
-FORMAT_STR_REMOVE_TRAILING_SEPARATOR = '@(shell_extension.FORMAT_STR_REMOVE_TRAILING_SEPARATOR)'
+FORMAT_STR_INVOKE_SCRIPT = '@(shell_extension.FORMAT_STR_INVOKE_SCRIPT)'  # noqa: E501
+FORMAT_STR_REMOVE_LEADING_SEPARATOR = '@(shell_extension.FORMAT_STR_REMOVE_LEADING_SEPARATOR)'  # noqa: E501
+FORMAT_STR_REMOVE_TRAILING_SEPARATOR = '@(shell_extension.FORMAT_STR_REMOVE_TRAILING_SEPARATOR)'  # noqa: E501
 
 DSV_TYPE_APPEND_NON_DUPLICATE = 'append-non-duplicate'
 DSV_TYPE_PREPEND_NON_DUPLICATE = 'prepend-non-duplicate'
diff --git a/colcon_core/task/python/build.py b/colcon_core/task/python/build.py
index 5dc89e442..a72c1a54e 100644
--- a/colcon_core/task/python/build.py
+++ b/colcon_core/task/python/build.py
@@ -8,7 +8,6 @@
 from pathlib import Path
 import shutil
 import sys
-from sys import executable
 
 from colcon_core.environment import create_environment_hooks
 from colcon_core.environment import create_environment_scripts
@@ -26,6 +25,12 @@
 
 logger = colcon_logger.getChild(__name__)
 
+_PYTHON_CMD = [
+    sys.executable,
+    '-W',
+    'ignore:setup.py install is deprecated',
+]
+
 
 def _get_install_scripts(path):
     setup_cfg_path = os.path.join(path, 'setup.cfg')
@@ -72,9 +77,12 @@ async def build(self, *, additional_hooks=None):  # noqa: D102
         python_lib = os.path.join(
             args.install_base, self._get_python_lib(args))
         os.makedirs(python_lib, exist_ok=True)
+        distutils_commands = os.path.join(
+            os.path.dirname(__file__), 'colcon_distutils_commands')
         # and being in the PYTHONPATH
         env = dict(env)
         env['PYTHONPATH'] = str(prefix_override) + os.pathsep + \
+            distutils_commands + os.pathsep + \
             python_lib + os.pathsep + env.get('PYTHONPATH', '')
         # coverage capture interferes with sitecustomize
         # See also: https://docs.python.org/3/library/site.html#module-site
@@ -92,7 +100,7 @@ async def build(self, *, additional_hooks=None):  # noqa: D102
 
             # invoke `setup.py install` step with lots of arguments
             # to avoid placing any files in the source space
-            cmd = [executable, 'setup.py']
+            cmd = _PYTHON_CMD + ['setup.py']
             if 'egg_info' in available_commands:
                 # `setup.py egg_info` requires the --egg-base to exist
                 os.makedirs(args.build_base, exist_ok=True)
@@ -139,8 +147,8 @@ async def build(self, *, additional_hooks=None):  # noqa: D102
             try:
                 # --editable causes this to skip creating/editing the
                 # easy-install.pth file
-                cmd = [
-                    executable, 'setup.py',
+                cmd = _PYTHON_CMD + [
+                    'setup.py',
                     'develop',
                     '--editable',
                     '--build-directory',
@@ -181,7 +189,7 @@ async def build(self, *, additional_hooks=None):  # noqa: D102
 
     async def _get_available_commands(self, path, env):
         output = await check_output(
-            [executable, 'setup.py', '--help-commands'], cwd=path, env=env)
+            _PYTHON_CMD + ['setup.py', '--help-commands'], cwd=path, env=env)
         commands = set()
         for line in output.splitlines():
             if not line.startswith(b'  '):
@@ -208,8 +216,8 @@ async def _undo_develop(self, pkg, args, env):
             args.build_base, '%s.egg-info' % pkg.name.replace('-', '_'))
         setup_py_build_space = os.path.join(args.build_base, 'setup.py')
         if os.path.exists(egg_info) and os.path.islink(setup_py_build_space):
-            cmd = [
-                executable, 'setup.py',
+            cmd = _PYTHON_CMD + [
+                'setup.py',
                 'develop',
                 '--uninstall', '--editable',
                 '--build-directory', os.path.join(args.build_base, 'build')
diff --git a/colcon_core/task/python/colcon_distutils_commands/__init__.py b/colcon_core/task/python/colcon_distutils_commands/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/colcon_core/task/python/colcon_distutils_commands/colcon_distutils_commands-0.0.0.dist-info/METADATA b/colcon_core/task/python/colcon_distutils_commands/colcon_distutils_commands-0.0.0.dist-info/METADATA
new file mode 100644
index 000000000..c97b22f57
--- /dev/null
+++ b/colcon_core/task/python/colcon_distutils_commands/colcon_distutils_commands-0.0.0.dist-info/METADATA
@@ -0,0 +1,3 @@
+Metadata-Version: 2.1
+Name: colcon_distutils_commands
+Version: 0.0.0
diff --git a/colcon_core/task/python/colcon_distutils_commands/colcon_distutils_commands-0.0.0.dist-info/entry_points.txt b/colcon_core/task/python/colcon_distutils_commands/colcon_distutils_commands-0.0.0.dist-info/entry_points.txt
new file mode 100644
index 000000000..cade4c272
--- /dev/null
+++ b/colcon_core/task/python/colcon_distutils_commands/colcon_distutils_commands-0.0.0.dist-info/entry_points.txt
@@ -0,0 +1,2 @@
+[distutils.commands]
+symlink_data = colcon_core.distutils.commands.symlink_data:symlink_data
diff --git a/colcon_core/task/python/test/__init__.py b/colcon_core/task/python/test/__init__.py
index 74a5ed2b8..a01ec4784 100644
--- a/colcon_core/task/python/test/__init__.py
+++ b/colcon_core/task/python/test/__init__.py
@@ -137,7 +137,7 @@ async def step(self):
         raise NotImplementedError()
 
 
-def get_python_testing_step_extensions():
+def get_python_testing_step_extensions(*, group_name=None):
     """
     Get the available Python testing step extensions.
 
@@ -145,8 +145,9 @@ def get_python_testing_step_extensions():
 
     :rtype: OrderedDict
     """
-    extensions = instantiate_extensions(
-        'colcon_core.python_testing', unique_instance=False)
+    if group_name is None:
+        group_name = 'colcon_core.python_testing'
+    extensions = instantiate_extensions(group_name, unique_instance=False)
     for name in list(extensions.keys()):
         extension = extensions[name]
         extension.STEP_TYPE = name
diff --git a/colcon_core/verb/__init__.py b/colcon_core/verb/__init__.py
index 640dc218f..49913ba16 100644
--- a/colcon_core/verb/__init__.py
+++ b/colcon_core/verb/__init__.py
@@ -48,7 +48,7 @@ def main(self, *, context):
         raise NotImplementedError()
 
 
-def get_verb_extensions():
+def get_verb_extensions(*, group_name=None):
     """
     Get the available verb extensions.
 
@@ -56,7 +56,9 @@ def get_verb_extensions():
 
     :rtype: OrderedDict
     """
-    extensions = instantiate_extensions(__name__)
+    if group_name is None:
+        group_name = __name__
+    extensions = instantiate_extensions(group_name)
     for name, extension in extensions.items():
         extension.VERB_NAME = name
     return order_extensions_by_name(extensions)
diff --git a/setup.cfg b/setup.cfg
index 640bf8b21..3928d02e7 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -145,17 +145,22 @@ colcon_core.verb =
     test = colcon_core.verb.test:TestVerb
 console_scripts =
     colcon = colcon_core.command:main
-distutils.commands =
-    symlink_data = colcon_core.distutils.commands.symlink_data:symlink_data
 pytest11 =
     colcon_core_warnings_stderr = colcon_core.pytest.hooks
 
 [options.package_data]
 colcon_core.shell.template = *.em
 colcon_core.task.python.template = *.em
+colcon_core.task.python.colcon_distutils_commands =
+    */METADATA
+    */entry_points.txt
 
 [flake8]
 import-order-style = google
+per-file-ignores =
+    colcon_core/distutils/__init__.py:A005
+    colcon_core/logging.py:A005
+    colcon_core/subprocess.py:A005
 
 [coverage:run]
 source = colcon_core
diff --git a/stdeb.cfg b/stdeb.cfg
index 3c5672b71..9eb1227d1 100644
--- a/stdeb.cfg
+++ b/stdeb.cfg
@@ -3,5 +3,6 @@ No-Python2:
 Depends3: python3-distlib, python3-empy (<4), python3-packaging, python3-pytest, python3-setuptools, python3 (>= 3.8) | python3-importlib-metadata
 Recommends3: python3-pytest-cov
 Suggests3: python3-pytest-repeat, python3-pytest-rerunfailures
+Replaces3: colcon
 Suite: focal jammy noble bookworm trixie
 X-Python3-Version: >= 3.6
diff --git a/test/spell_check.words b/test/spell_check.words
index d44f7a23b..18c2677bf 100644
--- a/test/spell_check.words
+++ b/test/spell_check.words
@@ -1,8 +1,10 @@
+addfinalizer
 addopts
 apache
 argparse
 asyncio
 autouse
+backported
 basepath
 bazqux
 blocklist
@@ -17,6 +19,7 @@ configparser
 contextlib
 coroutine
 coroutines
+cpython
 datetime
 debian
 debinfo
@@ -33,8 +36,10 @@ docstring
 executables
 exitstatus
 fdopen
+ffoo
 filterwarnings
 foobar
+fooo
 fromhex
 functools
 getcategory
@@ -50,6 +55,7 @@ importlib
 importorskip
 isatty
 iterdir
+itertools
 junit
 levelname
 libexec
@@ -110,6 +116,7 @@ setuptools
 shlex
 sigint
 sitecustomize
+skipif
 sloretz
 stacklevel
 staticmethod
@@ -127,6 +134,7 @@ testcase
 testsfailed
 testsuite
 thomas
+tmpdir
 todo
 traceback
 tryfirst
@@ -136,5 +144,6 @@ unittest
 unittests
 unlinking
 unrenamed
+usefixtures
 wildcards
 workaround
diff --git a/test/test_command.py b/test/test_command.py
index cdd547700..c8c6b6e9e 100644
--- a/test/test_command.py
+++ b/test/test_command.py
@@ -1,15 +1,18 @@
 # Copyright 2016-2018 Dirk Thomas
 # Licensed under the Apache License, Version 2.0
 
+import os
 import shutil
 import signal
 import sys
 from tempfile import mkdtemp
+from tempfile import TemporaryDirectory
 from unittest.mock import Mock
 from unittest.mock import patch
 
 from colcon_core.command import CommandContext
 from colcon_core.command import create_parser
+from colcon_core.command import get_prog_name
 from colcon_core.command import main
 from colcon_core.command import verb_main
 from colcon_core.environment_variable import EnvironmentVariable
@@ -82,6 +85,17 @@ def test_main():
             assert rc == signal.SIGINT
 
 
+def test_main_no_verbs_or_env():
+    with ExtensionPointContext():
+        with patch(
+            'colcon_core.command.load_extension_points',
+            return_value={},
+        ):
+            with pytest.raises(SystemExit) as e:
+                main(argv=['--help'])
+            assert e.value.code == 0
+
+
 def test_create_parser():
     with ExtensionPointContext():
         parser = create_parser('colcon_core.environment_variable')
@@ -140,3 +154,86 @@ def test_verb_main():
     assert logger.error.call_args[0][0].startswith(
         'command_name verb_name: custom error message\n')
     assert 'Exception: custom error message' in logger.error.call_args[0][0]
+
+
+def test_prog_name_module():
+    argv = [os.path.join('foo', 'bar', '__main__.py')]
+    with patch('colcon_core.command.sys.argv', argv):
+        # prog should be the module containing __main__.py
+        assert get_prog_name() == 'bar'
+
+
+def test_prog_name_on_path():
+    # use __file__ since we know it exists
+    argv = [__file__]
+    with patch('colcon_core.command.sys.argv', argv):
+        with patch(
+            'colcon_core.command.shutil.which',
+            return_value=__file__
+        ):
+            # prog should be shortened to the basename
+            assert get_prog_name() == 'test_command.py'
+
+
+def test_prog_name_not_on_path():
+    # use __file__ since we know it exists
+    argv = [__file__]
+    with patch('colcon_core.command.sys.argv', argv):
+        with patch('colcon_core.command.shutil.which', return_value=None):
+            # prog should remain unchanged
+            assert get_prog_name() == __file__
+
+
+def test_prog_name_different_on_path():
+    # use __file__ since we know it exists
+    argv = [__file__]
+    with patch('colcon_core.command.sys.argv', argv):
+        with patch(
+            'colcon_core.command.shutil.which',
+            return_value=sys.executable
+        ):
+            # prog should remain unchanged
+            assert get_prog_name() == __file__
+
+
+def test_prog_name_not_a_file():
+    # pick some file that doesn't actually exist on disk
+    no_such_file = os.path.join(__file__, 'foobar')
+    argv = [no_such_file]
+    with patch('colcon_core.command.sys.argv', argv):
+        with patch(
+            'colcon_core.command.shutil.which',
+            return_value=no_such_file
+        ):
+            # prog should remain unchanged
+            assert get_prog_name() == no_such_file
+
+
+@pytest.mark.skipif(sys.platform == 'win32', reason='Symlinks not supported.')
+def test_prog_name_symlink():
+    # use __file__ since we know it exists
+    with TemporaryDirectory(prefix='test_colcon_') as temp_dir:
+        linked_file = os.path.join(temp_dir, 'test_command.py')
+        os.symlink(__file__, linked_file)
+
+        argv = [linked_file]
+        with patch('colcon_core.command.sys.argv', argv):
+            with patch(
+                'colcon_core.command.shutil.which',
+                return_value=__file__
+            ):
+                # prog should be shortened to the basename
+                assert get_prog_name() == 'test_command.py'
+
+
+@pytest.mark.skipif(sys.platform != 'win32', reason='Only valid on Windows.')
+def test_prog_name_easy_install():
+    # use __file__ since we know it exists
+    argv = [__file__[:-3]]
+    with patch('colcon_core.command.sys.argv', argv):
+        with patch(
+            'colcon_core.command.shutil.which',
+            return_value=__file__
+        ):
+            # prog should be shortened to the basename
+            assert get_prog_name() == 'test_command'
diff --git a/test/test_extension_point.py b/test/test_extension_point.py
index 96e58a0de..f0fa80438 100644
--- a/test/test_extension_point.py
+++ b/test/test_extension_point.py
@@ -6,69 +6,78 @@
 from unittest.mock import DEFAULT
 from unittest.mock import patch
 
+try:
+    from importlib.metadata import Distribution
+except ImportError:
+    # TODO: Drop this with Python 3.7 support
+    from importlib_metadata import Distribution
+
+from colcon_core.environment_variable import EnvironmentVariable
+from colcon_core.extension_point import clear_entry_point_cache
 from colcon_core.extension_point import EntryPoint
 from colcon_core.extension_point import EXTENSION_POINT_GROUP_NAME
 from colcon_core.extension_point import get_all_extension_points
 from colcon_core.extension_point import get_extension_points
 from colcon_core.extension_point import load_extension_point
 from colcon_core.extension_point import load_extension_points
+from colcon_core.extension_point import override_blocklist_variable
 import pytest
 
 from .environment_context import EnvironmentContext
 
 
-Group1 = EntryPoint('group1', 'g1', EXTENSION_POINT_GROUP_NAME)
-Group2 = EntryPoint('group2', 'g2', EXTENSION_POINT_GROUP_NAME)
-ExtA = EntryPoint('extA', 'eA', Group1.name)
-ExtB = EntryPoint('extB', 'eB', Group1.name)
-
-
-class Dist():
-
-    version = '0.0.0'
+class _FakeDistribution(Distribution):
 
     def __init__(self, entry_points):
-        self.metadata = {'Name': f'dist-{id(self)}'}
-        self._entry_points = entry_points
+        entry_points_spec = []
+        for group_name, group_members in entry_points.items():
+            entry_points_spec.append(f'[{group_name}]')
+            for member_name, member_value in group_members:
+                entry_points_spec.append(f'{member_name} = {member_value}')
+            entry_points_spec.append('')
 
-    @property
-    def entry_points(self):
-        return list(self._entry_points)
+        self._files = {
+            'PKG-INFO': f'Name: dist-{id(self)}\nVersion: 0.0.0\n',
+            'entry_points.txt': '\n'.join(entry_points_spec) + '\n',
+        }
 
-    @property
-    def name(self):
-        return self.metadata['Name']
+    def read_text(self, filename):
+        return self._files.get(filename)
 
+    def locate_file(self, path):
+        return path
 
-def iter_entry_points(*, group=None):
-    if group == EXTENSION_POINT_GROUP_NAME:
-        return [Group1, Group2]
-    elif group == Group1.name:
-        return [ExtA, ExtB]
-    assert not group
-    return {
-        EXTENSION_POINT_GROUP_NAME: [Group1, Group2],
-        Group1.name: [ExtA, ExtB],
-    }
 
+def _distributions():
+    yield _FakeDistribution({
+        EXTENSION_POINT_GROUP_NAME: [('group1', 'g1')],
+        'group1': [('extA', 'eA'), ('extB', 'eB')],
+    })
+    yield _FakeDistribution({
+        EXTENSION_POINT_GROUP_NAME: [('group2', 'g2')],
+        'group2': [('extC', 'eC')],
+    })
+    yield _FakeDistribution({
+        'groupX': [('extD', 'eD')],
+    })
 
-def distributions():
-    return [
-        Dist([Group1, ExtA, ExtB]),
-        Dist([Group2, EntryPoint('extC', 'eC', Group2.name)]),
-        Dist([EntryPoint('extD', 'eD', 'groupX')]),
-    ]
+
+def _entry_points():
+    for dist in _distributions():
+        yield from dist.entry_points
 
 
 def test_all_extension_points():
     with patch(
         'colcon_core.extension_point.entry_points',
-        side_effect=iter_entry_points
+        side_effect=_entry_points
     ):
         with patch(
             'colcon_core.extension_point.distributions',
-            side_effect=distributions
+            side_effect=_distributions
         ):
+            clear_entry_point_cache()
+
             # successfully load a known entry point
             extension_points = get_all_extension_points()
             assert set(extension_points.keys()) == {
@@ -84,18 +93,20 @@ def test_extension_point_blocklist():
     # successful loading of extension point without a blocklist
     with patch(
         'colcon_core.extension_point.entry_points',
-        side_effect=iter_entry_points
+        side_effect=_entry_points
     ):
         with patch(
             'colcon_core.extension_point.distributions',
-            side_effect=distributions
+            side_effect=_distributions
         ):
+            clear_entry_point_cache()
             extension_points = get_extension_points('group1')
     assert 'extA' in extension_points.keys()
     extension_point = extension_points['extA']
     assert extension_point == 'eA'
 
     with patch.object(EntryPoint, 'load', return_value=None) as load:
+        clear_entry_point_cache()
         load_extension_point('extA', 'eA', 'group1')
         assert load.call_count == 1
 
@@ -104,12 +115,14 @@ def test_extension_point_blocklist():
         with EnvironmentContext(COLCON_EXTENSION_BLOCKLIST=os.pathsep.join([
             'group1.extB', 'group2.extC'])
         ):
+            clear_entry_point_cache()
             load_extension_point('extA', 'eA', 'group1')
         assert load.call_count == 1
 
         # entry point in a blocked group can't be loaded
         load.reset_mock()
         with EnvironmentContext(COLCON_EXTENSION_BLOCKLIST='group1'):
+            clear_entry_point_cache()
             with pytest.raises(RuntimeError) as e:
                 load_extension_point('extA', 'eA', 'group1')
             assert 'The entry point group name is listed in the environment ' \
@@ -120,6 +133,7 @@ def test_extension_point_blocklist():
         with EnvironmentContext(COLCON_EXTENSION_BLOCKLIST=os.pathsep.join([
             'group1.extA', 'group1.extB'])
         ):
+            clear_entry_point_cache()
             with pytest.raises(RuntimeError) as e:
                 load_extension_point('extA', 'eA', 'group1')
             assert 'The entry point name is listed in the environment ' \
@@ -127,6 +141,77 @@ def test_extension_point_blocklist():
         assert load.call_count == 0
 
 
+def test_extension_point_blocklist_override():
+    with patch.object(EntryPoint, 'load', return_value=None) as load:
+        clear_entry_point_cache()
+
+        my_extension_blocklist = EnvironmentVariable(
+            'MY_EXTENSION_BLOCKLIST', 'Foo bar baz')
+        override_blocklist_variable(my_extension_blocklist)
+
+        try:
+            # entry point in default blocklist variable can be loaded
+            load.reset_mock()
+            with EnvironmentContext(COLCON_EXTENSION_BLOCKLIST='group1'):
+                clear_entry_point_cache()
+                load_extension_point('extA', 'eA', 'group1')
+            assert load.call_count == 1
+
+            # entry point in custom blocklist variable can't be loaded
+            load.reset_mock()
+            with EnvironmentContext(MY_EXTENSION_BLOCKLIST='group1'):
+                clear_entry_point_cache()
+                with pytest.raises(RuntimeError) as e:
+                    load_extension_point('extA', 'eA', 'group1')
+                assert 'The entry point group name is listed in the ' \
+                    'environment variable' in str(e.value)
+            assert load.call_count == 0
+        finally:
+            override_blocklist_variable(None)
+
+        # entry point in default blocklist variable can no longer be loaded
+        load.reset_mock()
+        with EnvironmentContext(COLCON_EXTENSION_BLOCKLIST='group1'):
+            clear_entry_point_cache()
+            with pytest.raises(RuntimeError) as e:
+                load_extension_point('extA', 'eA', 'group1')
+            assert 'The entry point group name is listed in the ' \
+                'environment variable' in str(e.value)
+        assert load.call_count == 0
+
+
+def test_redefined_extension_point():
+    def _duped_distributions():
+        yield from _distributions()
+        yield _FakeDistribution({
+            'group2': [('extC', 'eC-prime')],
+        })
+
+    def _duped_entry_points():
+        for dist in _duped_distributions():
+            yield from dist.entry_points
+
+    with patch('colcon_core.extension_point.logger.error') as error:
+        with patch(
+            'colcon_core.extension_point.entry_points',
+            side_effect=_duped_entry_points
+        ):
+            with patch(
+                'colcon_core.extension_point.distributions',
+                side_effect=_duped_distributions
+            ):
+                clear_entry_point_cache()
+                extension_points = get_all_extension_points()
+                assert 'eC-prime' == extension_points['group2']['extC'][0]
+                assert error.call_count == 1
+
+                error.reset_mock()
+                clear_entry_point_cache()
+                extension_points = get_extension_points('group2')
+                assert 'eC-prime' == extension_points.get('extC')
+                assert error.call_count == 1
+
+
 def entry_point_load(self, *args, **kwargs):
     if self.name == 'exception':
         raise Exception('entry point raising exception')
diff --git a/test/test_feature_flags.py b/test/test_feature_flags.py
new file mode 100644
index 000000000..218b29773
--- /dev/null
+++ b/test/test_feature_flags.py
@@ -0,0 +1,106 @@
+# Copyright 2024 Open Source Robotics Foundation, Inc.
+# Licensed under the Apache License, Version 2.0
+
+import os
+from unittest.mock import patch
+
+from colcon_core.feature_flags import check_implemented_flags
+from colcon_core.feature_flags import FEATURE_FLAGS_ENVIRONMENT_VARIABLE
+from colcon_core.feature_flags import get_feature_flags
+from colcon_core.feature_flags import is_feature_flag_set
+import pytest
+
+
+_FLAGS_TO_TEST = (
+    ('foo',),
+    ('foo', 'foo'),
+    ('foo', ''),
+    ('', 'foo'),
+    ('', 'foo', ''),
+    ('foo', 'bar'),
+    ('bar', 'foo'),
+    ('bar', 'foo', 'baz'),
+)
+
+
+@pytest.fixture
+def feature_flags_value(request):
+    env = dict(os.environ)
+    if request.param is not None:
+        env[FEATURE_FLAGS_ENVIRONMENT_VARIABLE.name] = os.pathsep.join(
+            request.param)
+    else:
+        env.pop(FEATURE_FLAGS_ENVIRONMENT_VARIABLE.name, None)
+
+    mock_env = patch('colcon_core.feature_flags.os.environ', env)
+    request.addfinalizer(mock_env.stop)
+    mock_env.start()
+    return request.param
+
+
+@pytest.fixture
+def feature_flag_reports(request):
+    reported_uses = patch('colcon_core.feature_flags._REPORTED_USES', set())
+    request.addfinalizer(reported_uses.stop)
+    reported_uses.start()
+    return reported_uses
+
+
+@pytest.mark.parametrize(
+    'feature_flags_value',
+    _FLAGS_TO_TEST,
+    indirect=True)
+@pytest.mark.usefixtures('feature_flags_value', 'feature_flag_reports')
+def test_flag_is_set():
+    with patch('colcon_core.feature_flags.logger.warning') as warn:
+        assert is_feature_flag_set('foo')
+        assert warn.call_count == 2
+        assert is_feature_flag_set('foo')
+        assert warn.call_count == 2
+
+
+@pytest.mark.parametrize(
+    'feature_flags_value',
+    (None, *_FLAGS_TO_TEST),
+    indirect=True)
+@pytest.mark.usefixtures('feature_flags_value', 'feature_flag_reports')
+def test_flag_not_set():
+    with patch('colcon_core.feature_flags.logger.warning') as warn:
+        assert not is_feature_flag_set('')
+        assert not is_feature_flag_set('fo')
+        assert not is_feature_flag_set('oo')
+        assert not is_feature_flag_set('fooo')
+        assert not is_feature_flag_set('ffoo')
+        assert not is_feature_flag_set('qux')
+        assert warn.call_count == 0
+
+
+@pytest.mark.parametrize(
+    'feature_flags_value',
+    (None, *_FLAGS_TO_TEST),
+    indirect=True)
+@pytest.mark.usefixtures('feature_flags_value')
+def test_get_flags(feature_flags_value):
+    assert [
+        flag for flag in (feature_flags_value or ()) if flag
+    ] == get_feature_flags()
+
+
+@pytest.mark.parametrize('feature_flags_value', (('baz',),), indirect=True)
+@pytest.mark.usefixtures('feature_flags_value')
+def test_implemented():
+    with patch('colcon_core.feature_flags.IMPLEMENTED_FLAGS', {'foo'}):
+        with patch('colcon_core.feature_flags.logger.warning') as warn:
+            assert not is_feature_flag_set('bar')
+            assert warn.call_count == 0
+            assert is_feature_flag_set('baz')
+            assert warn.call_count == 2
+            assert is_feature_flag_set('foo')
+            assert warn.call_count == 2
+            check_implemented_flags()
+            assert warn.call_count == 2
+
+    with patch('colcon_core.feature_flags.IMPLEMENTED_FLAGS', {'baz'}):
+        with patch('colcon_core.feature_flags.logger.warning') as warn:
+            check_implemented_flags()
+            assert warn.call_count == 1
diff --git a/test/test_logging.py b/test/test_logging.py
index c67673872..350b04a60 100644
--- a/test/test_logging.py
+++ b/test/test_logging.py
@@ -2,8 +2,11 @@
 # Licensed under the Apache License, Version 2.0
 
 import logging
+from pathlib import Path
 from unittest.mock import Mock
 
+from colcon_core.logging import add_file_handler
+from colcon_core.logging import get_effective_console_level
 from colcon_core.logging import get_numeric_log_level
 from colcon_core.logging import set_logger_level_from_env
 import pytest
@@ -56,3 +59,46 @@ def test_get_numeric_log_level():
     with pytest.raises(ValueError) as e:
         get_numeric_log_level('-1')
     assert str(e.value).endswith('numeric log levels must be positive')
+
+
+def test_add_file_handler(tmpdir):
+    log_path = Path(tmpdir) / 'test_add_file_handler.log'
+    log_path.touch()
+    logger = logging.getLogger('test_add_file_handler')
+    try:
+        logger.setLevel(logging.WARN)
+        add_file_handler(logger, log_path)
+        assert logger.getEffectiveLevel() != logging.WARN
+        logger.info('test_add_file_handler')
+    finally:
+        for handler in logger.handlers:
+            logger.removeHandler(handler)
+            handler.close()
+
+    # check only that we logged SOMETHING to the file
+    assert log_path.stat().st_size > 10
+
+
+def test_get_effective_console_level(tmpdir):
+    logger = logging.getLogger('test_sync_console_log_level')
+
+    # no level set
+    level = get_effective_console_level(logger)
+    assert level == logger.getEffectiveLevel()
+
+    # change the level to ERROR
+    logger.setLevel(logging.ERROR)
+    level = get_effective_console_level(logger)
+    assert level == logger.getEffectiveLevel() == logging.ERROR
+
+    # after add_file_handler
+    log_path = Path(tmpdir) / 'test_add_file_handler.log'
+    log_path.touch()
+    try:
+        add_file_handler(logger, log_path)
+        level = get_effective_console_level(logger)
+        assert level == logging.ERROR
+    finally:
+        for handler in logger.handlers:
+            logger.removeHandler(handler)
+            handler.close()
diff --git a/test/test_package_descriptor.py b/test/test_package_descriptor.py
index 7c53c09a2..4baf7b247 100644
--- a/test/test_package_descriptor.py
+++ b/test/test_package_descriptor.py
@@ -1,6 +1,7 @@
 # Copyright 2016-2018 Dirk Thomas
 # Licensed under the Apache License, Version 2.0
 
+from collections import defaultdict
 import os
 from pathlib import Path
 
@@ -49,7 +50,8 @@ def test_get_dependencies():
     assert "'self'" in str(e.value)
 
 
-def test_get_recursive_dependencies():
+@pytest.fixture
+def recursive_dependencies():
     d = PackageDescriptor('/some/path')
     d.name = 'A'
     d.dependencies['build'].add('B')
@@ -70,6 +72,7 @@ def test_get_recursive_dependencies():
     d3.dependencies['build'].add('h')
     d3.dependencies['test'].add('G')
     d3.dependencies['test'].add('I')
+    d3.dependencies['test'].add('J')
 
     d4 = PackageDescriptor('/more/path')
     d4.name = 'G'
@@ -80,10 +83,35 @@ def test_get_recursive_dependencies():
     # circular dependencies should be ignored
     d5.dependencies['run'].add('A')
 
-    rec_deps = d.get_recursive_dependencies(
-        {d, d1, d2, d3, d4, d5},
+    d6 = PackageDescriptor('/paths/galore')
+    d6.name = 'J'
+
+    return d, {d, d1, d2, d3, d4, d5, d6}
+
+
+def test_get_recursive_dependencies(recursive_dependencies):
+    desc, all_descs = recursive_dependencies
+    rec_deps = desc.get_recursive_dependencies(
+        all_descs,
         direct_categories=('build', 'run'),
         recursive_categories=('run', 'test'))
+    assert rec_deps == {
+        # direct dependencies
+        'B',
+        # recursive dependencies
+        'F', 'G', 'I', 'J',
+    }
+
+
+def test_get_recursive_dependencies_map(recursive_dependencies):
+    recursive_categories = defaultdict(lambda: ('run', 'test'))
+    recursive_categories['run'] = ('run',)
+
+    desc, all_descs = recursive_dependencies
+    rec_deps = desc.get_recursive_dependencies(
+        all_descs,
+        direct_categories=('build', 'run'),
+        recursive_categories=recursive_categories)
     assert rec_deps == {
         # direct dependencies
         'B',