Repository: DataBiosphere/toil

Issue:
Good config file support for WDL and CWL runner options
The Toil config file should be able to apply to the WDL and CWL runner options.



┆Issue is synchronized with this [Jira Story](https://ucsc-cgl.atlassian.net/browse/TOIL-1422)
┆Issue Number: TOIL-1422


Discussion:

Patch:
diff --git a/contrib/mypy-stubs/configargparse/__init__.pyi b/contrib/mypy-stubs/configargparse/__init__.pyi
index e2c7b6b2..0c4451c5 100644
--- a/contrib/mypy-stubs/configargparse/__init__.pyi
+++ b/contrib/mypy-stubs/configargparse/__init__.pyi
@@ -1,3 +1,5 @@
 from .configargparse import ArgParser as ArgParser
 from .configargparse import ArgumentParser as ArgumentParser
+from .configargparse import Namespace as Namespace
+from .configargparse import SUPPRESS as SUPPRESS
 from .configargparse import YAMLConfigFileParser as YAMLConfigFileParser
diff --git a/contrib/mypy-stubs/configargparse/configargparse.pyi b/contrib/mypy-stubs/configargparse/configargparse.pyi
index 46a66df5..eda8ef53 100644
--- a/contrib/mypy-stubs/configargparse/configargparse.pyi
+++ b/contrib/mypy-stubs/configargparse/configargparse.pyi
@@ -1,5 +1,5 @@
 import argparse
-from typing import Any, OrderedDict, Sequence, TypeVar
+from typing import Any, List, OrderedDict, Sequence, TypeVar
 
 __all__ = ["ArgumentParser", "YAMLConfigFileParser", "ConfigFileParser"]
 _N = TypeVar("_N")
@@ -19,10 +19,17 @@ class YAMLConfigFileParser(ConfigFileParser):
 class ArgumentParser(argparse.ArgumentParser):
     @property
     def _config_file_parser(self) -> Any: ...
+    @property
+    def _default_config_files(self) -> List[Any]: ...
+    @property
+    def _ignore_unknown_config_file_keys(self) -> Any: ...
+
     def __init__(self, *args: Any, **kwargs: Any) -> None: ...
     # There may be a better way of type hinting this without a type: ignore, but mypy gets unhappy pretty much no matter what as the signatures for parse_args doesn't match with its superclass in argparse
     def parse_args(self, args: Sequence[str] | None = None, namespace: Namespace | None = None, config_file_contents: str | None = None, env_vars: Any = None) -> Namespace: ...  # type: ignore[override]
+    def parse_known_args(self, args: Sequence[str] | None = ..., namespace: Namespace | None = ..., config_file_contents: Any = ..., env_vars: Any = ..., ignore_help_args: bool | None = ...) -> tuple[Namespace, list[str]]: ...  # type: ignore[override]
     def get_source_to_settings_dict(self) -> OrderedDict[Any, Any]: ...
 
 Namespace = argparse.Namespace
 ArgParser = ArgumentParser
+SUPPRESS = argparse.SUPPRESS
diff --git a/docs/running/cliOptions.rst b/docs/running/cliOptions.rst
index 2e7324ae..2aec02a1 100644
--- a/docs/running/cliOptions.rst
+++ b/docs/running/cliOptions.rst
@@ -16,19 +16,45 @@ The Config File
 Instead of changing the arguments on the CLI, Toil offers support for using a configuration file (Note: Support for the
 configuration file and environmental variables require the use of ``configargparse``).
 
-To generate a default configuration file::
+Options will be applied with priority:
 
-    $ toil config [file].yaml
+  1. CLI
+
+  2. Environmental Variables
+
+  3. Config file values
+
+     a. Provided config file through CLI
+
+     b. Default config value in ``$HOME/.toil/default.yaml``
+
+  4. Defaults
+
+To generate a configuration file, run::
+
+    $ toil config [filename].yaml
+
+Then uncomment options as necessary and change/provide new values. 
 
 After editing the config file, make Toil take in the new options::
 
-    $ python example.py --config=[file].yaml
+    $ python example.py --config=[filename].yaml
+
+Alternatively, you can edit the default config file, which is located at ``$HOME/.toil/default.yaml``
+
+If CLI options are used in addition to the configuration file, the CLI options will overwrite the configuration file
+options. For example::
+
+    $ python example.py --config=[filename].yaml --maxNodes 20
+
+With configuration file::
+
+    ...
+    maxNodes=[10]
+    ...
 
-If CLI options are used in addition with the configuration file, the CLI options will overwrite the configuration file
-options::
+Will result in maxNodes=[20].
 
-    $ python example.py --config=[file].yaml --maxNodes 20
-    # maxNodes=[20] even though default maxNodes=[10]
 
 The Job Store
 -------------
diff --git a/requirements.txt b/requirements.txt
index b0e85191..c34aceb5 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -11,5 +11,6 @@ pytz>=2012
 enlighten>=1.5.2, <2
 configargparse>=1.7,<2
 ruamel.yaml>=0.15
+ruamel.yaml.string>=0.1.1
 pyyaml>=6,<7
 typing-extensions>=4.6.2, <5
diff --git a/src/toil/batchSystems/kubernetes.py b/src/toil/batchSystems/kubernetes.py
index 9a06105a..c9deef42 100644
--- a/src/toil/batchSystems/kubernetes.py
+++ b/src/toil/batchSystems/kubernetes.py
@@ -104,7 +104,8 @@ from toil.batchSystems.abstractBatchSystem import (EXIT_STATUS_UNAVAILABLE_VALUE
 from toil.batchSystems.cleanup_support import BatchSystemCleanupSupport
 from toil.batchSystems.contained_executor import pack_job
 from toil.batchSystems.options import OptionSetter
-from toil.common import SYS_MAX_SIZE, Config, Toil
+from toil.common import Config, Toil
+from toil.options.common import SYS_MAX_SIZE
 from toil.job import JobDescription, Requirer
 from toil.lib.conversions import human2bytes
 from toil.lib.misc import get_user_name, slow_down, utc_now
diff --git a/src/toil/batchSystems/options.py b/src/toil/batchSystems/options.py
index ebe07c9d..8f432ee9 100644
--- a/src/toil/batchSystems/options.py
+++ b/src/toil/batchSystems/options.py
@@ -80,7 +80,7 @@ def set_batchsystem_options(batch_system: Optional[str], set_option: OptionSette
 
 
 def add_all_batchsystem_options(parser: Union[ArgumentParser, _ArgumentGroup]) -> None:
-    from toil.common import SYS_MAX_SIZE
+    from toil.options.common import SYS_MAX_SIZE
 
     # Do the global cross-batch-system arguments
     parser.add_argument(
diff --git a/src/toil/batchSystems/singleMachine.py b/src/toil/batchSystems/singleMachine.py
index 68ab14ff..f0c85d9e 100644
--- a/src/toil/batchSystems/singleMachine.py
+++ b/src/toil/batchSystems/singleMachine.py
@@ -34,7 +34,8 @@ from toil.batchSystems.abstractBatchSystem import (EXIT_STATUS_UNAVAILABLE_VALUE
                                                    UpdatedBatchJobInfo)
 from toil.batchSystems.options import OptionSetter
 from toil.bus import ExternalBatchIdMessage
-from toil.common import SYS_MAX_SIZE, Config, Toil, make_open_interval_action
+from toil.common import Config, Toil
+from toil.options.common import SYS_MAX_SIZE, make_open_interval_action
 from toil.job import (AcceleratorRequirement,
                       JobDescription,
                       Requirer,
diff --git a/src/toil/common.py b/src/toil/common.py
index 56da0a1d..e5a38172 100644
--- a/src/toil/common.py
+++ b/src/toil/common.py
@@ -23,16 +23,15 @@ import tempfile
 import time
 import uuid
 import warnings
+
+from ruamel.yaml import YAML
+from ruamel.yaml.comments import CommentedMap
+from configargparse import ArgParser, YAMLConfigFileParser
 from argparse import (SUPPRESS,
-                      Action,
                       ArgumentDefaultsHelpFormatter,
                       ArgumentParser,
                       Namespace,
-                      _AppendAction,
-                      _ArgumentGroup,
-                      _StoreFalseAction,
-                      _StoreTrueAction)
-from distutils.util import strtobool
+                      _ArgumentGroup, Action, _StoreFalseAction, _StoreTrueAction, _AppendAction)
 from functools import lru_cache
 from types import TracebackType
 from typing import (IO,
@@ -53,9 +52,10 @@ from typing import (IO,
 from urllib.parse import urlparse, unquote, quote
 
 import requests
-from configargparse import ArgParser, YAMLConfigFileParser
-from ruamel.yaml import YAML
-from ruamel.yaml.comments import CommentedMap
+
+from toil.options.common import add_base_toil_options, JOBSTORE_HELP
+from toil.options.cwl import add_cwl_options
+from toil.options.wdl import add_wdl_options
 
 if sys.version_info >= (3, 8):
     from typing import Literal
@@ -74,17 +74,17 @@ from toil.bus import (ClusterDesiredSizeMessage,
                       MessageBus,
                       QueueSizeMessage)
 from toil.fileStores import FileID
-from toil.lib.aws import zone_to_region
+from toil.lib.aws import zone_to_region, build_tag_dict_from_env
 from toil.lib.compatibility import deprecated
-from toil.lib.conversions import bytes2human, human2bytes
-from toil.lib.io import AtomicFileCreate, try_path
+from toil.lib.io import try_path, AtomicFileCreate
 from toil.lib.retry import retry
 from toil.provisioners import (add_provisioner_options,
                                cluster_factory,
                                parse_node_types)
 from toil.realtimeLogger import RealtimeLogger
-from toil.statsAndLogging import add_logging_options, set_logging_from_options
-from toil.version import dockerRegistry, dockerTag, version
+from toil.statsAndLogging import (add_logging_options,
+                                  set_logging_from_options)
+from toil.version import dockerRegistry, dockerTag, version, baseVersion
 
 if TYPE_CHECKING:
     from toil.batchSystems.abstractBatchSystem import AbstractBatchSystem
@@ -97,11 +97,6 @@ if TYPE_CHECKING:
     from toil.provisioners.abstractProvisioner import AbstractProvisioner
     from toil.resource import ModuleDescriptor
 
-# aim to pack autoscaling jobs within a 30 minute block before provisioning a new node
-defaultTargetTime = 1800
-SYS_MAX_SIZE = 9223372036854775807
-# sys.max_size on 64 bit systems is 9223372036854775807, so that 32-bit systems
-# use the same number
 UUID_LENGTH = 32
 logger = logging.getLogger(__name__)
 
@@ -111,37 +106,8 @@ TOIL_HOME_DIR: str = os.path.join(os.path.expanduser("~"), ".toil")
 DEFAULT_CONFIG_FILE: str = os.path.join(TOIL_HOME_DIR, "default.yaml")
 
 
-def parse_jobstore(jobstore_uri: str) -> str:
-    """
-    Turn the jobstore string into it's corresponding URI
-    ex:
-    /path/to/jobstore -> file:/path/to/jobstore
-
-    If the jobstore string already is a URI, return the jobstore:
-    aws:/path/to/jobstore -> aws:/path/to/jobstore
-    :param jobstore_uri: string of the jobstore
-    :return: URI of the jobstore
-    """
-    name, rest = Toil.parseLocator(jobstore_uri)
-    if name == "file":
-        # We need to resolve relative paths early, on the leader, because the worker process
-        # may have a different working directory than the leader, e.g. under Mesos.
-        return Toil.buildLocator(name, os.path.abspath(rest))
-    else:
-        return jobstore_uri
-
-
-def parse_str_list(s: str) -> List[str]:
-    return [str(x) for x in s.split(",")]
-
-
-def parse_int_list(s: str) -> List[int]:
-    return [int(x) for x in s.split(",")]
-
-
 class Config:
     """Class to represent configuration operations for a toil workflow run."""
-
     logFile: Optional[str]
     logRotating: bool
     cleanWorkDir: str
@@ -219,7 +185,8 @@ class Config:
     # Parameters to limit service jobs, so preventing deadlock scheduling scenarios
     maxPreemptibleServiceJobs: int
     maxServiceJobs: int
-    deadlockWait: Union[float, int]
+    deadlockWait: Union[
+        float, int]
     deadlockCheckInterval: Union[float, int]
 
     # Resource requirements
@@ -230,7 +197,7 @@ class Config:
     # TODO: These names are generated programmatically in
     # Requirer._fetchRequirement so we can't use snake_case until we fix
     # that (and add compatibility getters/setters?)
-    defaultAccelerators: List["AcceleratorRequirement"]
+    defaultAccelerators: List['AcceleratorRequirement']
     maxCores: int
     maxMemory: int
     maxDisk: int
@@ -283,50 +250,12 @@ class Config:
     def set_from_default_config(self) -> None:
         # get defaults from a config file by simulating an argparse run
         # as Config often expects defaults to already be instantiated
-        if not os.path.exists(DEFAULT_CONFIG_FILE):
-            # The default config file did not appear to exist when we checked.
-            # It might exist now, though. Try creating it.
-            self.generate_config_file()
-        # Check on the config file to make sure it is sensible
-        config_status = os.stat(DEFAULT_CONFIG_FILE)
-        if config_status.st_size == 0:
-            # If we have an empty config file, someone has to manually delete
-            # it before we will work again.
-            raise RuntimeError(
-                f"Config file {DEFAULT_CONFIG_FILE} exists but is empty. Delete it! Stat says: {config_status}"
-            )
-        logger.debug("Loading %s byte default configuration", config_status.st_size)
-        try:
-            with open(DEFAULT_CONFIG_FILE, "r") as f:
-                yaml = YAML(typ="safe")
-                s = yaml.load(f)
-                logger.debug("Loaded default configuration: %s", json.dumps(s))
-        except:
-            # Something went wrong reading the default config, so dump its
-            # contents to the log.
-            logger.info(
-                "Configuration file contents: %s", open(DEFAULT_CONFIG_FILE, "r").read()
-            )
-            raise
         parser = ArgParser()
-        addOptions(parser, jobstore_as_flag=True)
-        ns = parser.parse_args(f"--config={DEFAULT_CONFIG_FILE}")
+        addOptions(parser, jobstore_as_flag=True, cwl=self.cwl)
+        # The parser already knows about the default config file
+        ns = parser.parse_args("")
         self.setOptions(ns)
 
-    def generate_config_file(self) -> None:
-        """
-        If the default config file does not exist, create it.
-
-        Raises an error if the default config file cannot be created.
-        Safe to run simultaneously in multiple processes. If this process runs
-        this function, it will always see the default config file existing with
-        parseable contents, even if other processes are racing to create it.
-
-        No process will see an empty or partially-written default config file.
-        """
-        check_and_create_toil_home_dir()
-        generate_config(DEFAULT_CONFIG_FILE)
-
     def prepare_start(self) -> None:
         """
         After options are set, prepare for initial start of workflow.
@@ -347,7 +276,8 @@ class Config:
     def setOptions(self, options: Namespace) -> None:
         """Creates a config object from the options object."""
 
-        def set_option(option_name: str, old_names: Optional[List[str]] = None) -> None:
+        def set_option(option_name: str,
+                       old_names: Optional[List[str]] = None) -> None:
             """
             Determine the correct value for the given option.
 
@@ -372,21 +302,15 @@ class Config:
                 for old_name in old_names:
                     # If the option is already set with the new name and not the old name
                     # prioritize the new name over the old name and break
-                    if (
-                        option_value is not None
-                        and option_value != []
-                        and option_value != {}
-                    ):
+                    if option_value is not None and option_value != [] and option_value != {}:
                         break
                     # Try all the old names in case user code is setting them
                     # in an options object.
                     # This does assume that all deprecated options have a default value of None
                     if getattr(options, old_name, None) is not None:
-                        warnings.warn(
-                            f"Using deprecated option field {old_name} to "
-                            f"provide value for config field {option_name}",
-                            DeprecationWarning,
-                        )
+                        warnings.warn(f'Using deprecated option field {old_name} to '
+                                      f'provide value for config field {option_name}',
+                                      DeprecationWarning)
                         option_value = getattr(options, old_name)
             if option_value is not None or not hasattr(self, option_name):
                 setattr(self, option_name, option_value)
@@ -401,14 +325,13 @@ class Config:
         set_option("stats")
         set_option("cleanWorkDir")
         set_option("clean")
-        set_option("clusterStats")
+        set_option('clusterStats')
         set_option("restart")
 
         # Batch system options
         set_option("batchSystem")
-        set_batchsystem_options(
-            None, cast("OptionSetter", set_option)
-        )  # None as that will make set_batchsystem_options iterate through all batch systems and set their corresponding values
+        set_batchsystem_options(None, cast("OptionSetter",
+                                           set_option))  # None as that will make set_batchsystem_options iterate through all batch systems and set their corresponding values
 
         # File store options
         set_option("symlinkImports", old_names=["linkImports"])
@@ -481,31 +404,25 @@ class Config:
 
         self.check_configuration_consistency()
 
+        logger.debug("Loaded configuration: %s", vars(options))
+
     def check_configuration_consistency(self) -> None:
         """Old checks that cannot be fit into an action class for argparse"""
         if self.writeLogs and self.writeLogsGzip:
-            raise ValueError(
-                "Cannot use both --writeLogs and --writeLogsGzip at the same time."
-            )
+            raise ValueError("Cannot use both --writeLogs and --writeLogsGzip at the same time.")
         if self.writeLogsFromAllJobs and not self.writeLogs and not self.writeLogsGzip:
-            raise ValueError(
-                "To enable --writeLogsFromAllJobs, either --writeLogs or --writeLogsGzip must be set."
-            )
+            raise ValueError("To enable --writeLogsFromAllJobs, either --writeLogs or --writeLogsGzip must be set.")
         for override in self.nodeStorageOverrides:
             tokens = override.split(":")
             if not any(tokens[0] in n[0] for n in self.nodeTypes):
-                raise ValueError(
-                    "Instance type in --nodeStorageOverrides must be in --nodeTypes"
-                )
+                raise ValueError("Instance type in --nodeStorageOverrides must be in --nodeTypes")
 
         if self.stats:
             if self.clean != "never" and self.clean is not None:
-                logger.warning(
-                    "Contradicting options passed: Clean flag is set to %s "
-                    "despite the stats flag requiring "
-                    "the jobStore to be intact at the end of the run. "
-                    "Setting clean to 'never'." % self.clean
-                )
+                logger.warning("Contradicting options passed: Clean flag is set to %s "
+                               "despite the stats flag requiring "
+                               "the jobStore to be intact at the end of the run. "
+                               "Setting clean to \'never\'." % self.clean)
             self.clean = "never"
 
     def __eq__(self, other: object) -> bool:
@@ -525,9 +442,36 @@ def check_and_create_toil_home_dir() -> None:
 
     dir_path = try_path(TOIL_HOME_DIR)
     if dir_path is None:
-        raise RuntimeError(
-            f"Cannot create or access Toil configuration directory {TOIL_HOME_DIR}"
-        )
+        raise RuntimeError(f"Cannot create or access Toil configuration directory {TOIL_HOME_DIR}")
+
+
+def check_and_create_default_config_file() -> None:
+    """
+    If the default config file does not exist, create it in the Toil home directory. Create the Toil home directory
+    if needed
+
+    Raises an error if the default config file cannot be created.
+    Safe to run simultaneously in multiple processes. If this process runs
+    this function, it will always see the default config file existing with
+    parseable contents, even if other processes are racing to create it.
+
+    No process will see an empty or partially-written default config file.
+    """
+    check_and_create_toil_home_dir()
+    # The default config file did not appear to exist when we checked.
+    # It might exist now, though. Try creating it.
+    check_and_create_config_file(DEFAULT_CONFIG_FILE)
+
+
+def check_and_create_config_file(filepath: str) -> None:
+    """
+    If the config file at the filepath does not exist, try creating it.
+    The parent directory should be created prior to calling this
+    :param filepath: path to config file
+    :return: None
+    """
+    if not os.path.exists(filepath):
+        generate_config(filepath)
 
 
 def generate_config(filepath: str) -> None:
@@ -536,6 +480,8 @@ def generate_config(filepath: str) -> None:
 
     Safe to run simultaneously in multiple processes. No process will see an
     empty or partially-written file at the given path.
+
+    Set include to "cwl" or "wdl" to include cwl options and wdl options respectfully
     """
     # this is placed in common.py rather than toilConfig.py to prevent circular imports
 
@@ -554,59 +500,74 @@ def generate_config(filepath: str) -> None:
     #   and --caching respectively
     # Skip StoreTrue and StoreFalse options that have opposite defaults as including it in the config would
     # override those defaults
-    deprecated_or_redundant_options = (
-        "help",
-        "config",
-        "logCritical",
-        "logDebug",
-        "logError",
-        "logInfo",
-        "logOff",
-        "logWarning",
-        "linkImports",
-        "noLinkImports",
-        "moveExports",
-        "noMoveExports",
-        "enableCaching",
-        "disableCaching",
-    )
+    deprecated_or_redundant_options = ("help", "config", "logCritical", "logDebug", "logError", "logInfo", "logOff",
+                                       "logWarning", "linkImports", "noLinkImports", "moveExports", "noMoveExports",
+                                       "enableCaching", "disableCaching", "version")
 
-    parser = ArgParser(YAMLConfigFileParser())
-    addOptions(parser, jobstore_as_flag=True)
-
-    data = CommentedMap()  # to preserve order
-    group_title_key: Dict[str, str] = dict()
-    for action in parser._actions:
-        if any(
-            s.replace("-", "") in deprecated_or_redundant_options
-            for s in action.option_strings
-        ):
-            continue
-        # if action is StoreFalse and default is True then don't include
-        if isinstance(action, _StoreFalseAction) and action.default is True:
-            continue
-        # if action is StoreTrue and default is False then don't include
-        if isinstance(action, _StoreTrueAction) and action.default is False:
-            continue
-
-        option_string = (
-            action.option_strings[0]
-            if action.option_strings[0].find("--") != -1
-            else action.option_strings[1]
-        )
-        option = option_string[2:]
+    def create_config_dict_from_parser(parser: ArgumentParser) -> CommentedMap:
+        """
+        Creates a CommentedMap of the config file output from a given parser. This will put every parser action and it's
+        default into the output
+
+        :param parser: parser to generate from
+        :return: CommentedMap of what to put into the config file
+        """
+        data = CommentedMap()  # to preserve order
+        group_title_key: Dict[str, str] = dict()
+        for action in parser._actions:
+            if any(s.replace("-", "") in deprecated_or_redundant_options for s in action.option_strings):
+                continue
+            # if action is StoreFalse and default is True then don't include
+            if isinstance(action, _StoreFalseAction) and action.default is True:
+                continue
+            # if action is StoreTrue and default is False then don't include
+            if isinstance(action, _StoreTrueAction) and action.default is False:
+                continue
+
+            if len(action.option_strings) == 0:
+                continue
 
-        default = action.default
+            option_string = action.option_strings[0] if action.option_strings[0].find("--") != -1 else \
+                action.option_strings[1]
+            option = option_string[2:]
 
-        data[option] = default
+            default = action.default
 
-        # store where each argparse group starts
-        group_title = action.container.title  # type: ignore[attr-defined]
-        group_title_key.setdefault(group_title, option)
+            data[option] = default
 
-    # add comment for when each argparse group starts
-    for group_title, key in group_title_key.items():
-        data.yaml_set_comment_before_after_key(key, group_title)
+            # store where each argparse group starts
+            group_title = action.container.title  # type: ignore[attr-defined]
+            group_title_key.setdefault(group_title, option)
+
+        # add comment for when each argparse group starts
+        for group_title, key in group_title_key.items():
+            data.yaml_set_comment_before_after_key(key, group_title)
+
+        return data
+
+    all_data = []
+
+    parser = ArgParser(YAMLConfigFileParser())
+    add_base_toil_options(parser, jobstore_as_flag=True, cwl=False)
+    toil_base_data = create_config_dict_from_parser(parser)
+
+    toil_base_data.yaml_set_start_comment("This is the configuration file for Toil. To set an option, uncomment an "
+                                          "existing option and set its value. The current values are the defaults. "
+                                          "If the default configuration file is outdated, it can be refreshed with "
+                                          "`toil config ~/.toil/default.yaml`.\n\nBASE TOIL OPTIONS\n")
+    all_data.append(toil_base_data)
+
+    parser = ArgParser(YAMLConfigFileParser())
+    add_cwl_options(parser)
+    toil_cwl_data = create_config_dict_from_parser(parser)
+    toil_cwl_data.yaml_set_start_comment("\nTOIL CWL RUNNER OPTIONS")
+    all_data.append(toil_cwl_data)
+
+    parser = ArgParser(YAMLConfigFileParser())
+    add_wdl_options(parser)
+    toil_wdl_data = create_config_dict_from_parser(parser)
+    toil_wdl_data.yaml_set_start_comment("\nTOIL WDL RUNNER OPTIONS")
+    all_data.append(toil_wdl_data)
 
     # Now we need to put the config file in place at filepath.
     # But someone else may have already created a file at that path, or may be
@@ -618,29 +579,20 @@ def generate_config(filepath: str) -> None:
     # there to begin with).
     with AtomicFileCreate(filepath) as temp_path:
         with open(temp_path, "w") as f:
-            yaml = YAML()
-            yaml.dump(data, f)
-
-
-JOBSTORE_HELP = (
-    "The location of the job store for the workflow.  "
-    "A job store holds persistent information about the jobs, stats, and files in a "
-    "workflow. If the workflow is run with a distributed batch system, the job "
-    "store must be accessible by all worker nodes. Depending on the desired "
-    "job store implementation, the location should be formatted according to "
-    "one of the following schemes:\n\n"
-    "file:<path> where <path> points to a directory on the file systen\n\n"
-    "aws:<region>:<prefix> where <region> is the name of an AWS region like "
-    "us-west-2 and <prefix> will be prepended to the names of any top-level "
-    "AWS resources in use by job store, e.g. S3 buckets.\n\n "
-    "google:<project_id>:<prefix> TODO: explain\n\n"
-    "For backwards compatibility, you may also specify ./foo (equivalent to "
-    "file:./foo or just file:foo) or /bar (equivalent to file:/bar)."
-)
+            f.write("config_version: 1.0\n")
+            yaml = YAML(typ=['rt', 'string'])
+            for data in all_data:
+                if "config_version" in data:
+                    del data["config_version"]
+                for line in yaml.dump_to_string(data).split("\n"):  # type: ignore[attr-defined]
+                    if line:
+                        f.write("#")
+                    f.write(line)
+                    f.write("\n")
 
 
 def parser_with_common_options(
-    provisioner_options: bool = False, jobstore_option: bool = True
+        provisioner_options: bool = False, jobstore_option: bool = True
 ) -> ArgParser:
     parser = ArgParser(prog="Toil", formatter_class=ArgumentDefaultsHelpFormatter)
 
@@ -648,983 +600,110 @@ def parser_with_common_options(
         add_provisioner_options(parser)
 
     if jobstore_option:
-        parser.add_argument("jobStore", type=str, help=JOBSTORE_HELP)
+        parser.add_argument('jobStore', type=str, help=JOBSTORE_HELP)
 
     # always add these
     add_logging_options(parser)
-    parser.add_argument("--version", action="version", version=version)
-    parser.add_argument(
-        "--tempDirRoot",
-        dest="tempDirRoot",
-        type=str,
-        default=tempfile.gettempdir(),
-        help="Path to where temporary directory containing all temp files are created, "
-        "by default generates a fresh tmp dir with 'tempfile.gettempdir()'.",
-    )
+    parser.add_argument("--version", action='version', version=version)
+    parser.add_argument("--tempDirRoot", dest="tempDirRoot", type=str, default=tempfile.gettempdir(),
+                        help="Path to where temporary directory containing all temp files are created, "
+                             "by default generates a fresh tmp dir with 'tempfile.gettempdir()'.")
     return parser
 
 
-# This is kept in the outer scope as multiple batchsystem files use this
-def make_open_interval_action(
-    min: Union[int, float], max: Optional[Union[int, float]] = None
-) -> Type[Action]:
-    """
-    Returns an argparse action class to check if the input is within the given half-open interval.
-    ex:
-    Provided value to argparse must be within the interval [min, max)
-    Types of min and max must be the same (max may be None)
-
-    :param min: float/int
-    :param max: optional float/int
-    :return: argparse action class
+def addOptions(parser: ArgumentParser, jobstore_as_flag: bool = False, cwl: bool = False, wdl: bool = False) -> None:
     """
+    Add all Toil command line options to a parser.
 
-    class IntOrFloatOpenAction(Action):
-        def __call__(
-            self, parser: Any, namespace: Any, values: Any, option_string: Any = None
-        ) -> None:
-            if isinstance(min, int):
-                if max is not None:  # for mypy
-                    assert isinstance(max, int)
-                func = iC(min, max)
-            else:
-                func = fC(min, max)
-            try:
-                if not func(values):
-                    raise parser.error(
-                        f"{option_string} ({values}) must be within the range: [{min}, {'infinity' if max is None else max})"
-                    )
-            except AssertionError:
-                raise RuntimeError(
-                    f"The {option_string} option has an invalid value: {values}"
-                )
-            setattr(namespace, self.dest, values)
-
-    return IntOrFloatOpenAction
-
-
-def addOptions(
-    parser: ArgumentParser, jobstore_as_flag: bool = False, cwl: bool = False
-) -> None:
-    """
-    Add Toil command line options to a parser.
+    Support for config files if using configargparse. This will also check and set up the default config file.
 
-    Support for config files.
+    :param jobstore_as_flag: make the job store option a --jobStore flag instead of a required jobStore positional argument.
 
-    :param config: If specified, take defaults from the given Config.
+    :param cwl: Whether CWL options are expected. If so, CWL options won't be suppressed.
 
-    :param jobstore_as_flag: make the job store option a --jobStore flag instead of a required jobStore positional argument.
+    :param wdl:  Whether WDL options are expected. If so, WDL options won't be suppressed.
     """
-
+    if cwl and wdl:
+        raise RuntimeError("CWL and WDL cannot both be true at the same time when adding options.")
     if not (isinstance(parser, ArgumentParser) or isinstance(parser, _ArgumentGroup)):
         raise ValueError(
-            f"Unanticipated class: {parser.__class__}.  Must be: argparse.ArgumentParser or ArgumentGroup."
-        )
+            f"Unanticipated class: {parser.__class__}.  Must be: argparse.ArgumentParser or ArgumentGroup.")
 
     if isinstance(parser, ArgParser):
         # in case the user passes in their own configargparse instance instead of calling getDefaultArgumentParser()
         # this forces configargparser to process the config file in YAML rather than in it's own format
         parser._config_file_parser = YAMLConfigFileParser()  # type: ignore[misc]
+        parser._default_config_files = [DEFAULT_CONFIG_FILE]  # type: ignore[misc]
     else:
         # configargparse advertises itself as a drag and drop replacement, and running the normal argparse ArgumentParser
         # through this code still seems to work (with the exception of --config and environmental variables)
-        warnings.warn(
-            f"Using deprecated library argparse for options parsing."
-            f"This will not parse config files or use environment variables."
-            f"Use configargparse instead or call Job.Runner.getDefaultArgumentParser()",
-            DeprecationWarning,
-        )
-
-    opt_strtobool = lambda b: b if b is None else bool(strtobool(b))
-    convert_bool = lambda b: bool(strtobool(b))
-    add_logging_options(parser)
-    parser.register("type", "bool", parseBool)  # Custom type for arg=True/False.
-
-    # Core options
-    core_options = parser.add_argument_group(
-        title="Toil core options.",
-        description="Options to specify the location of the Toil workflow and "
-        "turn on stats collation about the performance of jobs.",
-    )
-    if jobstore_as_flag:
-        core_options.add_argument(
-            "--jobstore",
-            "--jobStore",
-            dest="jobStore",
-            type=parse_jobstore,
-            default=None,
-            help=JOBSTORE_HELP,
-        )
-    else:
-        core_options.add_argument("jobStore", type=parse_jobstore, help=JOBSTORE_HELP)
-
-    class WorkDirAction(Action):
-        """
-        Argparse action class to check that the provided --workDir exists
-        """
-
-        def __call__(
-            self, parser: Any, namespace: Any, values: Any, option_string: Any = None
-        ) -> None:
-            workDir = values
-            if workDir is not None:
-                workDir = os.path.abspath(workDir)
-                if not os.path.exists(workDir):
-                    raise RuntimeError(
-                        f"The path provided to --workDir ({workDir}) does not exist."
-                    )
-
-                if len(workDir) > 80:
-                    logger.warning(
-                        f'Length of workDir path "{workDir}" is {len(workDir)} characters.  '
-                        f"Consider setting a shorter path with --workPath or setting TMPDIR to something "
-                        f'like "/tmp" to avoid overly long paths.'
-                    )
-            setattr(namespace, self.dest, workDir)
-
-    class CoordinationDirAction(Action):
-        """
-        Argparse action class to check that the provided --coordinationDir exists
-        """
-
-        def __call__(
-            self, parser: Any, namespace: Any, values: Any, option_string: Any = None
-        ) -> None:
-            coordination_dir = values
-            if coordination_dir is not None:
-                coordination_dir = os.path.abspath(coordination_dir)
-                if not os.path.exists(coordination_dir):
-                    raise RuntimeError(
-                        f"The path provided to --coordinationDir ({coordination_dir}) does not exist."
-                    )
-            setattr(namespace, self.dest, coordination_dir)
-
-    def make_closed_interval_action(
-        min: Union[int, float], max: Optional[Union[int, float]] = None
-    ) -> Type[Action]:
-        """
-        Returns an argparse action class to check if the input is within the given half-open interval.
-        ex:
-        Provided value to argparse must be within the interval [min, max]
-
-        :param min: int/float
-        :param max: optional int/float
-        :return: argparse action
-        """
-
-        class ClosedIntOrFloatAction(Action):
-            def __call__(
-                self,
-                parser: Any,
-                namespace: Any,
-                values: Any,
-                option_string: Any = None,
-            ) -> None:
-                def is_within(x: Union[int, float]) -> bool:
-                    if max is None:
-                        return min <= x
-                    else:
-                        return min <= x <= max
-
-                try:
-                    if not is_within(values):
-                        raise parser.error(
-                            f"{option_string} ({values}) must be within the range: [{min}, {'infinity' if max is None else max}]"
-                        )
-                except AssertionError:
-                    raise RuntimeError(
-                        f"The {option_string} option has an invalid value: {values}"
-                    )
-                setattr(namespace, self.dest, values)
-
-        return ClosedIntOrFloatAction
-
-    core_options.add_argument(
-        "--workDir",
-        dest="workDir",
-        default=None,
-        env_var="TOIL_WORKDIR",
-        action=WorkDirAction,
-        help="Absolute path to directory where temporary files generated during the Toil "
-        "run should be placed. Standard output and error from batch system jobs "
-        "(unless --noStdOutErr is set) will be placed in this directory. A cache directory "
-        "may be placed in this directory. Temp files and folders will be placed in a "
-        "directory toil-<workflowID> within workDir. The workflowID is generated by "
-        "Toil and will be reported in the workflow logs. Default is determined by the "
-        "variables (TMPDIR, TEMP, TMP) via mkdtemp. This directory needs to exist on "
-        "all machines running jobs; if capturing standard output and error from batch "
-        "system jobs is desired, it will generally need to be on a shared file system. "
-        "When sharing a cache between containers on a host, this directory must be "
-        "shared between the containers.",
-    )
-    core_options.add_argument(
-        "--coordinationDir",
-        dest="coordination_dir",
-        default=None,
-        env_var="TOIL_COORDINATION_DIR",
-        action=CoordinationDirAction,
-        help="Absolute path to directory where Toil will keep state and lock files."
-        "When sharing a cache between containers on a host, this directory must be "
-        "shared between the containers.",
-    )
-    core_options.add_argument(
-        "--noStdOutErr",
-        dest="noStdOutErr",
-        default=False,
-        action="store_true",
-        help="Do not capture standard output and error from batch system jobs.",
-    )
-    core_options.add_argument(
-        "--stats",
-        dest="stats",
-        default=False,
-        action="store_true",
-        help="Records statistics about the toil workflow to be used by 'toil stats'.",
-    )
-    clean_choices = ["always", "onError", "never", "onSuccess"]
-    core_options.add_argument(
-        "--clean",
-        dest="clean",
-        choices=clean_choices,
-        default="onSuccess",
-        help=f"Determines the deletion of the jobStore upon completion of the program.  "
-        f"Choices: {clean_choices}.  The --stats option requires information from the "
-        f"jobStore upon completion so the jobStore will never be deleted with that flag.  "
-        f"If you wish to be able to restart the run, choose 'never' or 'onSuccess'.  "
-        f"Default is 'never' if stats is enabled, and 'onSuccess' otherwise.",
-    )
-    core_options.add_argument(
-        "--cleanWorkDir",
-        dest="cleanWorkDir",
-        choices=clean_choices,
-        default="always",
-        help=f"Determines deletion of temporary worker directory upon completion of a job.  "
-        f"Choices: {clean_choices}.  Default = always.  WARNING: This option should be "
-        f"changed for debugging only.  Running a full pipeline with this option could "
-        f"fill your disk with excessive intermediate data.",
-    )
-    core_options.add_argument(
-        "--clusterStats",
-        dest="clusterStats",
-        nargs="?",
-        action="store",
-        default=None,
-        const=os.getcwd(),
-        help="If enabled, writes out JSON resource usage statistics to a file.  "
-        "The default location for this file is the current working directory, but an "
-        "absolute path can also be passed to specify where this file should be written. "
-        "This options only applies when using scalable batch systems.",
-    )
-
-    # Restarting the workflow options
-    restart_options = parser.add_argument_group(
-        title="Toil options for restarting an existing workflow.",
-        description="Allows the restart of an existing workflow",
-    )
-    restart_options.add_argument(
-        "--restart",
-        dest="restart",
-        default=False,
-        action="store_true",
-        help="If --restart is specified then will attempt to restart existing workflow "
-        "at the location pointed to by the --jobStore option. Will raise an exception "
-        "if the workflow does not exist",
-    )
-
-    # Batch system options
-    batchsystem_options = parser.add_argument_group(
-        title="Toil options for specifying the batch system.",
-        description="Allows the specification of the batch system.",
-    )
-    add_all_batchsystem_options(batchsystem_options)
-
-    # File store options
-    file_store_options = parser.add_argument_group(
-        title="Toil options for configuring storage.",
-        description="Allows configuring Toil's data storage.",
-    )
-    link_imports = file_store_options.add_mutually_exclusive_group()
-    link_imports_help = (
-        "When using a filesystem based job store, CWL input files are by default symlinked in.  "
-        "Setting this option to True instead copies the files into the job store, which may protect "
-        "them from being modified externally.  When set to False, as long as caching is enabled, "
-        "Toil will protect the file automatically by changing the permissions to read-only."
-        "default=%(default)s"
-    )
-    link_imports.add_argument(
-        "--symlinkImports",
-        dest="symlinkImports",
-        type=convert_bool,
-        default=True,
-        help=link_imports_help,
-    )
-    move_exports = file_store_options.add_mutually_exclusive_group()
-    move_exports_help = (
-        "When using a filesystem based job store, output files are by default moved to the "
-        "output directory, and a symlink to the moved exported file is created at the initial "
-        "location.  Setting this option to True instead copies the files into the output directory.  "
-        "Applies to filesystem-based job stores only."
-        "default=%(default)s"
-    )
-    move_exports.add_argument(
-        "--moveOutputs",
-        dest="moveOutputs",
-        type=convert_bool,
-        default=False,
-        help=move_exports_help,
-    )
-
-    caching = file_store_options.add_mutually_exclusive_group()
-    caching_help = "Enable or disable caching for your workflow, specifying this overrides default from job store"
-    caching.add_argument(
-        "--caching", dest="caching", type=opt_strtobool, default=None, help=caching_help
-    )
-    # default is None according to PR 4299, seems to be generated at runtime
-
-    # Auto scaling options
-    autoscaling_options = parser.add_argument_group(
-        title="Toil options for autoscaling the cluster of worker nodes.",
-        description="Allows the specification of the minimum and maximum number of nodes in an autoscaled cluster, "
-        "as well as parameters to control the level of provisioning.",
-    )
-    provisioner_choices = ["aws", "gce", None]
-
-    # TODO: Better consolidate this provisioner arg and the one in provisioners/__init__.py?
-    autoscaling_options.add_argument(
-        "--provisioner",
-        "-p",
-        dest="provisioner",
-        choices=provisioner_choices,
-        default=None,
-        help=f"The provisioner for cluster auto-scaling.  This is the main Toil "
-        f"'--provisioner' option, and defaults to None for running on single "
-        f"machine and non-auto-scaling batch systems.  The currently supported "
-        f"choices are {provisioner_choices}.  The default is %(default)s.",
-    )
-    autoscaling_options.add_argument(
-        "--nodeTypes",
-        default=[],
-        dest="nodeTypes",
-        type=parse_node_types,
-        action="extend",
-        help="Specifies a list of comma-separated node types, each of which is "
-        "composed of slash-separated instance types, and an optional spot "
-        "bid set off by a colon, making the node type preemptible. Instance "
-        "types may appear in multiple node types, and the same node type "
-        "may appear as both preemptible and non-preemptible.\n"
-        "Valid argument specifying two node types:\n"
-        "\tc5.4xlarge/c5a.4xlarge:0.42,t2.large\n"
-        "Node types:\n"
-        "\tc5.4xlarge/c5a.4xlarge:0.42 and t2.large\n"
-        "Instance types:\n"
-        "\tc5.4xlarge, c5a.4xlarge, and t2.large\n"
-        "Semantics:\n"
-        "\tBid $0.42/hour for either c5.4xlarge or c5a.4xlarge instances,\n"
-        "\ttreated interchangeably, while they are available at that price,\n"
-        "\tand buy t2.large instances at full price.\n"
-        "default=%(default)s",
-    )
-
-    class NodeExtendAction(_AppendAction):
-        """
-        argparse Action class to remove the default value on first call, and act as an extend action after
-        """
-
-        # with action=append/extend, the argparse default is always prepended to the option
-        # so make the CLI have priority by rewriting the option on the first run
-        def __init__(self, option_strings: Any, dest: Any, **kwargs: Any):
-            super().__init__(option_strings, dest, **kwargs)
-            self.is_default = True
-
-        def __call__(
-            self, parser: Any, namespace: Any, values: Any, option_string: Any = None
-        ) -> None:
-            if self.is_default:
-                setattr(namespace, self.dest, values)
-                self.is_default = False
-            else:
-                super().__call__(parser, namespace, values, option_string)
-
-    autoscaling_options.add_argument(
-        "--maxNodes",
-        default=[10],
-        dest="maxNodes",
-        type=parse_int_list,
-        action=NodeExtendAction,
-        help=f"Maximum number of nodes of each type in the cluster, if using autoscaling, "
-        f"provided as a comma-separated list.  The first value is used as a default "
-        f"if the list length is less than the number of nodeTypes.  "
-        f"default=%(default)s",
-    )
-    autoscaling_options.add_argument(
-        "--minNodes",
-        default=[0],
-        dest="minNodes",
-        type=parse_int_list,
-        action=NodeExtendAction,
-        help="Mininum number of nodes of each type in the cluster, if using "
-        "auto-scaling.  This should be provided as a comma-separated list of the "
-        "same length as the list of node types. default=%(default)s",
-    )
-    autoscaling_options.add_argument(
-        "--targetTime",
-        dest="targetTime",
-        default=defaultTargetTime,
-        type=int,
-        action=make_closed_interval_action(0),
-        help=f"Sets how rapidly you aim to complete jobs in seconds. Shorter times mean "
-        f"more aggressive parallelization. The autoscaler attempts to scale up/down "
-        f"so that it expects all queued jobs will complete within targetTime "
-        f"seconds.  default=%(default)s",
-    )
-    autoscaling_options.add_argument(
-        "--betaInertia",
-        dest="betaInertia",
-        default=0.1,
-        type=float,
-        action=make_closed_interval_action(0.0, 0.9),
-        help=f"A smoothing parameter to prevent unnecessary oscillations in the number "
-        f"of provisioned nodes. This controls an exponentially weighted moving "
-        f"average of the estimated number of nodes. A value of 0.0 disables any "
-        f"smoothing, and a value of 0.9 will smooth so much that few changes will "
-        f"ever be made.  Must be between 0.0 and 0.9. default=%(default)s",
-    )
-    autoscaling_options.add_argument(
-        "--scaleInterval",
-        dest="scaleInterval",
-        default=60,
-        type=int,
-        help=f"The interval (seconds) between assessing if the scale of "
-        f"the cluster needs to change. default=%(default)s",
-    )
-    autoscaling_options.add_argument(
-        "--preemptibleCompensation",
-        "--preemptableCompensation",
-        dest="preemptibleCompensation",
-        default=0.0,
-        type=float,
-        action=make_closed_interval_action(0.0, 1.0),
-        help=f"The preference of the autoscaler to replace preemptible nodes with "
-        f"non-preemptible nodes, when preemptible nodes cannot be started for some "
-        f"reason. This value must be between 0.0 and 1.0, inclusive.  "
-        f"A value of 0.0 disables such "
-        f"compensation, a value of 0.5 compensates two missing preemptible nodes "
-        f"with a non-preemptible one. A value of 1.0 replaces every missing "
-        f"pre-emptable node with a non-preemptible one. default=%(default)s",
-    )
-    autoscaling_options.add_argument(
-        "--nodeStorage",
-        dest="nodeStorage",
-        default=50,
-        type=int,
-        help="Specify the size of the root volume of worker nodes when they are launched "
-        "in gigabytes. You may want to set this if your jobs require a lot of disk "
-        f"space.  (default=%(default)s).",
-    )
-    autoscaling_options.add_argument(
-        "--nodeStorageOverrides",
-        dest="nodeStorageOverrides",
-        default=[],
-        type=parse_str_list,
-        action="extend",
-        help="Comma-separated list of nodeType:nodeStorage that are used to override "
-        "the default value from --nodeStorage for the specified nodeType(s).  "
-        "This is useful for heterogeneous jobs where some tasks require much more "
-        "disk than others.",
-    )
-
-    autoscaling_options.add_argument(
-        "--metrics",
-        dest="metrics",
-        default=False,
-        type=convert_bool,
-        help="Enable the prometheus/grafana dashboard for monitoring CPU/RAM usage, "
-        "queue size, and issued jobs.",
-    )
-    autoscaling_options.add_argument(
-        "--assumeZeroOverhead",
-        dest="assume_zero_overhead",
-        default=False,
-        type=convert_bool,
-        help="Ignore scheduler and OS overhead and assume jobs can use every last byte "
-        "of memory and disk on a node when autoscaling.",
-    )
-
-    # Parameters to limit service jobs / detect service deadlocks
-    service_options = parser.add_argument_group(
-        title="Toil options for limiting the number of service jobs and detecting service deadlocks",
-        description="Allows the specification of the maximum number of service jobs in a cluster.  By keeping "
-        "this limited we can avoid nodes occupied with services causing deadlocks.",
-    )
-    service_options.add_argument(
-        "--maxServiceJobs",
-        dest="maxServiceJobs",
-        default=SYS_MAX_SIZE,
-        type=int,
-        help=SUPPRESS
-        if cwl
-        else f"The maximum number of service jobs that can be run concurrently, "
-        f"excluding service jobs running on preemptible nodes.  "
-        f"default=%(default)s",
-    )
-    service_options.add_argument(
-        "--maxPreemptibleServiceJobs",
-        dest="maxPreemptibleServiceJobs",
-        default=SYS_MAX_SIZE,
-        type=int,
-        help=SUPPRESS
-        if cwl
-        else f"The maximum number of service jobs that can run concurrently on "
-        f"preemptible nodes.  default=%(default)s",
-    )
-    service_options.add_argument(
-        "--deadlockWait",
-        dest="deadlockWait",
-        default=60,
-        type=int,
-        help=SUPPRESS
-        if cwl
-        else f"Time, in seconds, to tolerate the workflow running only the same service "
-        f"jobs, with no jobs to use them, before declaring the workflow to be "
-        f"deadlocked and stopping.  default=%(default)s",
-    )
-    service_options.add_argument(
-        "--deadlockCheckInterval",
-        dest="deadlockCheckInterval",
-        default=30,
-        type=int,
-        help=SUPPRESS
-        if cwl
-        else "Time, in seconds, to wait between checks to see if the workflow is stuck "
-        "running only service jobs, with no jobs to use them. Should be shorter "
-        "than --deadlockWait. May need to be increased if the batch system cannot "
-        "enumerate running jobs quickly enough, or if polling for running jobs is "
-        "placing an unacceptable load on a shared cluster.  "
-        f"default=%(default)s",
-    )
-
-    # Resource requirements
-    resource_options = parser.add_argument_group(
-        title="Toil options for cores/memory requirements.",
-        description="The options to specify default cores/memory requirements (if not specified by the jobs "
-        "themselves), and to limit the total amount of memory/cores requested from the batch system.",
-    )
-    resource_help_msg = (
-        "The {} amount of {} to request for a job.  "
-        "Only applicable to jobs that do not specify an explicit value for this requirement.  "
-        "{}.  "
-        "Default is {}."
-    )
-    cpu_note = "Fractions of a core (for example 0.1) are supported on some batch systems [mesos, single_machine]"
-    disk_mem_note = "Standard suffixes like K, Ki, M, Mi, G or Gi are supported"
-    accelerators_note = (
-        "Each accelerator specification can have a type (gpu [default], nvidia, amd, cuda, rocm, opencl, "
-        "or a specific model like nvidia-tesla-k80), and a count [default: 1]. If both a type and a count "
-        "are used, they must be separated by a colon. If multiple types of accelerators are "
-        "used, the specifications are separated by commas"
-    )
-
-    h2b = lambda x: human2bytes(str(x))
-
-    resource_options.add_argument(
-        "--defaultMemory",
-        dest="defaultMemory",
-        default="2.0 Gi",
-        type=h2b,
-        action=make_open_interval_action(1),
-        help=resource_help_msg.format(
-            "default", "memory", disk_mem_note, bytes2human(2147483648)
-        ),
-    )
-    resource_options.add_argument(
-        "--defaultCores",
-        dest="defaultCores",
-        default=1,
-        metavar="FLOAT",
-        type=float,
-        action=make_open_interval_action(1.0),
-        help=resource_help_msg.format("default", "cpu", cpu_note, str(1)),
-    )
-    resource_options.add_argument(
-        "--defaultDisk",
-        dest="defaultDisk",
-        default="2.0 Gi",
-        metavar="INT",
-        type=h2b,
-        action=make_open_interval_action(1),
-        help=resource_help_msg.format(
-            "default", "disk", disk_mem_note, bytes2human(2147483648)
-        ),
-    )
-    resource_options.add_argument(
-        "--defaultAccelerators",
-        dest="defaultAccelerators",
-        default=[],
-        metavar="ACCELERATOR[,ACCELERATOR...]",
-        type=parse_accelerator_list,
-        action="extend",
-        help=resource_help_msg.format("default", "accelerators", accelerators_note, []),
-    )
-    resource_options.add_argument(
-        "--defaultPreemptible",
-        "--defaultPreemptable",
-        dest="defaultPreemptible",
-        metavar="BOOL",
-        type=convert_bool,
-        nargs="?",
-        const=True,
-        default=False,
-        help="Make all jobs able to run on preemptible (spot) nodes by default.",
-    )
-    resource_options.add_argument(
-        "--maxCores",
-        dest="maxCores",
-        default=SYS_MAX_SIZE,
-        metavar="INT",
-        type=int,
-        action=make_open_interval_action(1),
-        help=resource_help_msg.format("max", "cpu", cpu_note, str(SYS_MAX_SIZE)),
-    )
-    resource_options.add_argument(
-        "--maxMemory",
-        dest="maxMemory",
-        default=SYS_MAX_SIZE,
-        metavar="INT",
-        type=h2b,
-        action=make_open_interval_action(1),
-        help=resource_help_msg.format(
-            "max", "memory", disk_mem_note, bytes2human(SYS_MAX_SIZE)
-        ),
-    )
-    resource_options.add_argument(
-        "--maxDisk",
-        dest="maxDisk",
-        default=SYS_MAX_SIZE,
-        metavar="INT",
-        type=h2b,
-        action=make_open_interval_action(1),
-        help=resource_help_msg.format(
-            "max", "disk", disk_mem_note, bytes2human(SYS_MAX_SIZE)
-        ),
-    )
-
-    # Retrying/rescuing jobs
-    job_options = parser.add_argument_group(
-        title="Toil options for rescuing/killing/restarting jobs.",
-        description="The options for jobs that either run too long/fail or get lost (some batch systems have issues!).",
-    )
-    job_options.add_argument(
-        "--retryCount",
-        dest="retryCount",
-        default=1,
-        type=int,
-        action=make_open_interval_action(0),
-        help=f"Number of times to retry a failing job before giving up and "
-        f"labeling job failed. default={1}",
-    )
-    job_options.add_argument(
-        "--enableUnlimitedPreemptibleRetries",
-        "--enableUnlimitedPreemptableRetries",
-        dest="enableUnlimitedPreemptibleRetries",
-        type=convert_bool,
-        default=False,
-        help="If set, preemptible failures (or any failure due to an instance getting "
-        "unexpectedly terminated) will not count towards job failures and --retryCount.",
-    )
-    job_options.add_argument(
-        "--doubleMem",
-        dest="doubleMem",
-        type=convert_bool,
-        default=False,
-        help="If set, batch jobs which die to reaching memory limit on batch schedulers "
-        "will have their memory doubled and they will be retried. The remaining "
-        "retry count will be reduced by 1. Currently supported by LSF.",
-    )
-    job_options.add_argument(
-        "--maxJobDuration",
-        dest="maxJobDuration",
-        default=SYS_MAX_SIZE,
-        type=int,
-        action=make_open_interval_action(1),
-        help=f"Maximum runtime of a job (in seconds) before we kill it (this is a lower bound, "
-        f"and the actual time before killing the job may be longer).  "
-        f"default=%(default)s",
-    )
-    job_options.add_argument(
-        "--rescueJobsFrequency",
-        dest="rescueJobsFrequency",
-        default=60,
-        type=int,
-        action=make_open_interval_action(1),
-        help=f"Period of time to wait (in seconds) between checking for missing/overlong jobs, "
-        f"that is jobs which get lost by the batch system. Expert parameter.  "
-        f"default=%(default)s",
-    )
-
-    # Log management options
-    log_options = parser.add_argument_group(
-        title="Toil log management options.",
-        description="Options for how Toil should manage its logs.",
-    )
-    log_options.add_argument(
-        "--maxLogFileSize",
-        dest="maxLogFileSize",
-        default=64000,
-        type=h2b,
-        action=make_open_interval_action(1),
-        help=f"The maximum size of a job log file to keep (in bytes), log files larger than "
-        f"this will be truncated to the last X bytes. Setting this option to zero will "
-        f"prevent any truncation. Setting this option to a negative value will truncate "
-        f"from the beginning.  Default={bytes2human(64000)}",
-    )
-    log_options.add_argument(
-        "--writeLogs",
-        dest="writeLogs",
-        nargs="?",
-        action="store",
-        default=None,
-        const=os.getcwd(),
-        help="Write worker logs received by the leader into their own files at the specified "
-        "path. Any non-empty standard output and error from failed batch system jobs will "
-        "also be written into files at this path.  The current working directory will be "
-        "used if a path is not specified explicitly. Note: By default only the logs of "
-        "failed jobs are returned to leader. Set log level to 'debug' or enable "
-        "'--writeLogsFromAllJobs' to get logs back from successful jobs, and adjust "
-        "'maxLogFileSize' to control the truncation limit for worker logs.",
-    )
-    log_options.add_argument(
-        "--writeLogsGzip",
-        dest="writeLogsGzip",
-        nargs="?",
-        action="store",
-        default=None,
-        const=os.getcwd(),
-        help="Identical to --writeLogs except the logs files are gzipped on the leader.",
-    )
-    log_options.add_argument(
-        "--writeLogsFromAllJobs",
-        dest="writeLogsFromAllJobs",
-        type=convert_bool,
-        default=False,
-        help="Whether to write logs from all jobs (including the successful ones) without "
-        "necessarily setting the log level to 'debug'. Ensure that either --writeLogs "
-        "or --writeLogsGzip is set if enabling this option.",
-    )
-    log_options.add_argument(
-        "--writeMessages",
-        dest="write_messages",
-        default=None,
-        type=lambda x: None if x is None else os.path.abspath(x),
-        help="File to send messages from the leader's message bus to.",
-    )
-    log_options.add_argument(
-        "--realTimeLogging",
-        dest="realTimeLogging",
-        type=convert_bool,
-        default=False,
-        help="Enable real-time logging from workers to leader",
-    )
-
-    # Misc options
-    misc_options = parser.add_argument_group(
-        title="Toil miscellaneous options.", description="Everything else."
-    )
-    misc_options.add_argument(
-        "--disableChaining",
-        dest="disableChaining",
-        type=convert_bool,
-        default=False,
-        help="Disables chaining of jobs (chaining uses one job's resource allocation "
-        "for its successor job if possible).",
-    )
-    misc_options.add_argument(
-        "--disableJobStoreChecksumVerification",
-        dest="disableJobStoreChecksumVerification",
-        default=False,
-        type=convert_bool,
-        help="Disables checksum verification for files transferred to/from the job store.  "
-        "Checksum verification is a safety check to ensure the data is not corrupted "
-        "during transfer. Currently only supported for non-streaming AWS files.",
-    )
-
-    class SSEKeyAction(Action):
-        def __call__(
-            self, parser: Any, namespace: Any, values: Any, option_string: Any = None
-        ) -> None:
-            if values is not None:
-                sse_key = values
-                if sse_key is None:
-                    return
-                with open(sse_key) as f:
-                    assert (
-                        len(f.readline().rstrip()) == 32
-                    ), "SSE key appears to be invalid."
-            setattr(namespace, self.dest, values)
-
-    misc_options.add_argument(
-        "--sseKey",
-        dest="sseKey",
-        default=None,
-        action=SSEKeyAction,
-        help="Path to file containing 32 character key to be used for server-side encryption on "
-        "awsJobStore or googleJobStore. SSE will not be used if this flag is not passed.",
-    )
-
-    # yaml.safe_load is being deprecated, this is the suggested workaround
-    def yaml_safe_load(stream: Any) -> Any:
-        yaml = YAML(typ="safe", pure=True)
-        d = yaml.load(stream)
-        if isinstance(d, dict):
-            # this means the argument was a dictionary and is valid yaml (for configargparse)
-            return d
-        else:
-            # this means the argument is likely in it's string format (for CLI)
-            return parseSetEnv(parse_str_list(stream))
-
-    class ExtendActionDict(Action):
+        warnings.warn(f'Using deprecated library argparse for options parsing.'
+                      f'This will not parse config files or use environment variables.'
+                      f'Use configargparse instead or call Job.Runner.getDefaultArgumentParser()',
+                      DeprecationWarning)
+
+    check_and_create_default_config_file()
+    # Check on the config file to make sure it is sensible
+    config_status = os.stat(DEFAULT_CONFIG_FILE)
+    if config_status.st_size == 0:
+        # If we have an empty config file, someone has to manually delete
+        # it before we will work again.
+        raise RuntimeError(
+            f"Config file {DEFAULT_CONFIG_FILE} exists but is empty. Delete it! Stat says: {config_status}")
+    try:
+        with open(DEFAULT_CONFIG_FILE, "r") as f:
+            yaml = YAML(typ="safe")
+            s = yaml.load(f)
+            logger.debug("Initialized default configuration: %s", json.dumps(s))
+    except:
+        # Something went wrong reading the default config, so dump its
+        # contents to the log.
+        logger.info("Configuration file contents: %s", open(DEFAULT_CONFIG_FILE, 'r').read())
+        raise
+
+    # Add base toil options
+    add_base_toil_options(parser, jobstore_as_flag, cwl)
+    # Add CWL and WDL options
+    # This is done so the config file can hold all available options
+    add_cwl_options(parser, suppress=not cwl)
+    add_wdl_options(parser, suppress=not wdl)
+
+    def check_arguments(typ: str) -> None:
         """
-        Argparse action class to implement the action="extend" functionality on dictionaries
+        Check that the other opposing runner's options are not on the command line.
+        Ex: if the parser is supposed to be a CWL parser, ensure that WDL commands are not on the command line
+        :param typ: string of either "cwl" or "wdl" to specify which runner to check against
+        :return: None, raise parser error if option is found
         """
-
-        def __call__(
-            self, parser: Any, namespace: Any, values: Any, option_string: Any = None
-        ) -> None:
-            items = getattr(namespace, self.dest, None)
-            assert (
-                items is not None
-            )  # for mypy. This should never be None, esp. if called in setEnv
-            # note: this will overwrite existing entries
-            items.update(values)
-
-    misc_options.add_argument(
-        "--setEnv",
-        "-e",
-        metavar="NAME=VALUE or NAME",
-        dest="environment",
-        default={},
-        type=yaml_safe_load,
-        action=ExtendActionDict,
-        help="Set an environment variable early on in the worker. If VALUE is null, it will "
-        "be looked up in the current environment. Independently of this option, the worker "
-        "will try to emulate the leader's environment before running a job, except for "
-        "some variables known to vary across systems.  Using this option, a variable can "
-        "be injected into the worker process itself before it is started.",
-    )
-    misc_options.add_argument(
-        "--servicePollingInterval",
-        dest="servicePollingInterval",
-        default=60.0,
-        type=float,
-        action=make_open_interval_action(0.0),
-        help=f"Interval of time service jobs wait between polling for the existence of the "
-        f"keep-alive flag.  Default: {60.0}",
-    )
-    misc_options.add_argument(
-        "--forceDockerAppliance",
-        dest="forceDockerAppliance",
-        type=convert_bool,
-        default=False,
-        help="Disables sanity checking the existence of the docker image specified by "
-        "TOIL_APPLIANCE_SELF, which Toil uses to provision mesos for autoscaling.",
-    )
-    misc_options.add_argument(
-        "--statusWait",
-        dest="statusWait",
-        type=int,
-        default=3600,
-        help="Seconds to wait between reports of running jobs.",
-    )
-    misc_options.add_argument(
-        "--disableProgress",
-        dest="disableProgress",
-        type=convert_bool,
-        default=False,
-        help="Disables the progress bar shown when standard error is a terminal.",
-    )
-    # If using argparse instead of configargparse, this should just not parse when calling parse_args()
-    # default config value is set to none as defaults should already be populated at config init
-    misc_options.add_argument(
-        "--config",
-        dest="config",
-        is_config_file_arg=True,
-        default=None,
-        help="Get options from a config file.",
-    )
-
-    # Debug options
-    debug_options = parser.add_argument_group(
-        title="Toil debug options.",
-        description="Debug options for finding problems or helping with testing.",
-    )
-    debug_options.add_argument(
-        "--debugWorker",
-        dest="debugWorker",
-        default=False,
-        action="store_true",
-        help="Experimental no forking mode for local debugging.  Specifically, workers "
-        "are not forked and stderr/stdout are not redirected to the log.",
-    )
-    debug_options.add_argument(
-        "--disableWorkerOutputCapture",
-        dest="disableWorkerOutputCapture",
-        default=False,
-        action="store_true",
-        help="Let worker output go to worker's standard out/error instead of per-job logs.",
-    )
-    debug_options.add_argument(
-        "--badWorker",
-        dest="badWorker",
-        default=0.0,
-        type=float,
-        action=make_closed_interval_action(0.0, 1.0),
-        help=f"For testing purposes randomly kill --badWorker proportion of jobs using "
-        f"SIGKILL.  default={0.0}",
-    )
-    debug_options.add_argument(
-        "--badWorkerFailInterval",
-        dest="badWorkerFailInterval",
-        default=0.01,
-        type=float,
-        action=make_open_interval_action(0.0),
-        help=f"When killing the job pick uniformly within the interval from 0.0 to "
-        f"--badWorkerFailInterval seconds after the worker starts.  "
-        f"default={0.01}",
-    )
-
-    # All deprecated options:
-
-    # These are deprecated in favor of a simpler option
-    #   ex: noLinkImports and linkImports can be simplified into a single link_imports argument
-    link_imports.add_argument(
-        "--noLinkImports", dest="linkImports", action="store_false", help=SUPPRESS
-    )
-    link_imports.add_argument(
-        "--linkImports", dest="linkImports", action="store_true", help=SUPPRESS
-    )
-    link_imports.set_defaults(linkImports=None)
-
-    move_exports.add_argument(
-        "--moveExports", dest="moveExports", action="store_true", help=SUPPRESS
-    )
-    move_exports.add_argument(
-        "--noMoveExports", dest="moveExports", action="store_false", help=SUPPRESS
-    )
-    link_imports.set_defaults(moveExports=None)
-
-    # dest is set to enableCaching to not conflict with the current --caching destination
-    caching.add_argument(
-        "--disableCaching", dest="enableCaching", action="store_false", help=SUPPRESS
-    )
-    caching.set_defaults(disableCaching=None)
-
-
-def parseBool(val: str) -> bool:
-    if val.lower() in ["true", "t", "yes", "y", "on", "1"]:
-        return True
-    elif val.lower() in ["false", "f", "no", "n", "off", "0"]:
-        return False
-    else:
-        raise RuntimeError('Could not interpret "%s" as a boolean value' % val)
+        check_parser = ArgParser()
+        if typ == "wdl":
+            add_cwl_options(check_parser)
+        if typ == "cwl":
+            add_wdl_options(check_parser)
+        for action in check_parser._actions:
+            action.default = SUPPRESS
+        other_options, _ = check_parser.parse_known_args(sys.argv[1:], ignore_help_args=True)
+        if len(vars(other_options)) != 0:
+            raise parser.error(f"{'WDL' if typ == 'cwl' else 'CWL'} options are not allowed on the command line.")
+
+    # if cwl is set, format the namespace for cwl and check that wdl options are not set on the command line
+    if cwl:
+        parser.add_argument("cwltool", type=str, help="CWL file to run.")
+        parser.add_argument("cwljob", nargs="*", help="Input file or CWL options. If CWL workflow takes an input, "
+                                                      "the name of the input can be used as an option. "
+                                                      "For example: \"%(prog)s workflow.cwl --file1 file\". "
+                                                      "If an input has the same name as a Toil option, pass '--' before it.")
+        check_arguments(typ="cwl")
+
+    # if wdl is set, format the namespace for wdl and check that cwl options are not set on the command line
+    if wdl:
+        parser.add_argument("wdl_uri", type=str,
+                            help="WDL document URI")
+        parser.add_argument("inputs_uri", type=str, nargs='?',
+                            help="WDL input JSON URI")
+        parser.add_argument("--input", "-i", dest="inputs_uri", type=str,
+                            help="WDL input JSON URI")
+        check_arguments(typ="wdl")
 
 
 @lru_cache(maxsize=None)
@@ -1646,20 +725,15 @@ def getNodeID() -> str:
                 with open(idSourceFile) as inp:
                     nodeID = inp.readline().strip()
             except OSError:
-                logger.warning(
-                    f"Exception when trying to read ID file {idSourceFile}.  "
-                    f"Will try next method to get node ID.",
-                    exc_info=True,
-                )
+                logger.warning(f"Exception when trying to read ID file {idSourceFile}.  "
+                               f"Will try next method to get node ID.", exc_info=True)
             else:
                 if len(nodeID.split()) == 1:
                     logger.debug(f"Obtained node ID {nodeID} from file {idSourceFile}")
                     break
                 else:
-                    logger.warning(
-                        f"Node ID {nodeID} from file {idSourceFile} contains spaces.  "
-                        f"Will try next method to get node ID."
-                    )
+                    logger.warning(f"Node ID {nodeID} from file {idSourceFile} contains spaces.  "
+                                   f"Will try next method to get node ID.")
     else:
         nodeIDs = []
         for i_call in range(2):
@@ -1673,22 +747,18 @@ def getNodeID() -> str:
             if nodeIDs[0] == nodeIDs[1]:
                 nodeID = nodeIDs[0]
             else:
-                logger.warning(
-                    f"Different node IDs {nodeIDs} received from repeated calls to uuid.getnode().  "
-                    f"You should use another method to generate node ID."
-                )
+                logger.warning(f"Different node IDs {nodeIDs} received from repeated calls to uuid.getnode().  "
+                               f"You should use another method to generate node ID.")
 
             logger.debug(f"Obtained node ID {nodeID} from uuid.getnode()")
     if not nodeID:
-        logger.warning(
-            "Failed to generate stable node ID, returning empty string. If you see this message with a "
-            "work dir on a shared file system when using workers running on multiple nodes, you might "
-            "experience cryptic job failures"
-        )
-    if len(nodeID.replace("-", "")) < UUID_LENGTH:
+        logger.warning("Failed to generate stable node ID, returning empty string. If you see this message with a "
+                       "work dir on a shared file system when using workers running on multiple nodes, you might "
+                       "experience cryptic job failures")
+    if len(nodeID.replace('-', '')) < UUID_LENGTH:
         # Some platforms (Mac) give us not enough actual hex characters.
         # Repeat them so the result is convertable to a uuid.UUID
-        nodeID = nodeID.replace("-", "")
+        nodeID = nodeID.replace('-', '')
         num_repeats = UUID_LENGTH // len(nodeID) + 1
         nodeID = nodeID * num_repeats
         nodeID = nodeID[:UUID_LENGTH]
@@ -1701,7 +771,6 @@ class Toil(ContextManager["Toil"]):
 
     Specifically the batch system, job store, and its configuration.
     """
-
     config: Config
     _jobStore: "AbstractJobStore"
     _batchSystem: "AbstractBatchSystem"
@@ -1760,10 +829,10 @@ class Toil(ContextManager["Toil"]):
         return self
 
     def __exit__(
-        self,
-        exc_type: Optional[Type[BaseException]],
-        exc_val: Optional[BaseException],
-        exc_tb: Optional[TracebackType],
+            self,
+            exc_type: Optional[Type[BaseException]],
+            exc_val: Optional[BaseException],
+            exc_tb: Optional[TracebackType],
     ) -> Literal[False]:
         """
         Clean up after a workflow invocation.
@@ -1771,32 +840,24 @@ class Toil(ContextManager["Toil"]):
         Depending on the configuration, delete the job store.
         """
         try:
-            if (
-                exc_type is not None
-                and self.config.clean == "onError"
-                or exc_type is None
-                and self.config.clean == "onSuccess"
-                or self.config.clean == "always"
-            ):
+            if (exc_type is not None and self.config.clean == "onError" or
+                    exc_type is None and self.config.clean == "onSuccess" or
+                    self.config.clean == "always"):
+
                 try:
                     if self.config.restart and not self._inRestart:
                         pass
                     else:
                         self._jobStore.destroy()
-                        logger.info(
-                            "Successfully deleted the job store: %s"
-                            % str(self._jobStore)
-                        )
+                        logger.info("Successfully deleted the job store: %s" % str(self._jobStore))
                 except:
-                    logger.info(
-                        "Failed to delete the job store: %s" % str(self._jobStore)
-                    )
+                    logger.info("Failed to delete the job store: %s" % str(self._jobStore))
                     raise
         except Exception as e:
             if exc_type is None:
                 raise
             else:
-                logger.exception("The following error was raised during clean up:")
+                logger.exception('The following error was raised during clean up:')
         self._inContextManager = False
         self._inRestart = False
         return False  # let exceptions through
@@ -1819,10 +880,8 @@ class Toil(ContextManager["Toil"]):
         self._jobStore.write_leader_node_id()
 
         if self.config.restart:
-            raise ToilRestartException(
-                "A Toil workflow can only be started once. Use "
-                "Toil.restart() to resume it."
-            )
+            raise ToilRestartException('A Toil workflow can only be started once. Use '
+                                       'Toil.restart() to resume it.')
 
         self._batchSystem = self.createBatchSystem(self.config)
         self._setupAutoDeployment(rootJob.getUserScript())
@@ -1835,7 +894,7 @@ class Toil(ContextManager["Toil"]):
             # a shared file, where we can find and unpickle it at the end of the workflow.
             # Unpickling the promise will automatically substitute the promise for the actual
             # return value.
-            with self._jobStore.write_shared_file_stream("rootJobReturnValue") as fH:
+            with self._jobStore.write_shared_file_stream('rootJobReturnValue') as fH:
                 rootJob.prepareForPromiseRegistration(self._jobStore)
                 promise = rootJob.rv()
                 pickle.dump(promise, fH, protocol=pickle.HIGHEST_PROTOCOL)
@@ -1863,18 +922,15 @@ class Toil(ContextManager["Toil"]):
         self._jobStore.write_leader_node_id()
 
         if not self.config.restart:
-            raise ToilRestartException(
-                "A Toil workflow must be initiated with Toil.start(), " "not restart()."
-            )
+            raise ToilRestartException('A Toil workflow must be initiated with Toil.start(), '
+                                       'not restart().')
 
         from toil.job import JobException
-
         try:
             self._jobStore.load_root_job()
         except JobException:
             logger.warning(
-                "Requested restart but the workflow has already been completed; allowing exports to rerun."
-            )
+                'Requested restart but the workflow has already been completed; allowing exports to rerun.')
             return self._jobStore.get_root_job_return_value()
 
         self._batchSystem = self.createBatchSystem(self.config)
@@ -1893,14 +949,12 @@ class Toil(ContextManager["Toil"]):
         if self.config.provisioner is None:
             self._provisioner = None
         else:
-            self._provisioner = cluster_factory(
-                provisioner=self.config.provisioner,
-                clusterName=None,
-                zone=None,  # read from instance meta-data
-                nodeStorage=self.config.nodeStorage,
-                nodeStorageOverrides=self.config.nodeStorageOverrides,
-                sseKey=self.config.sseKey,
-            )
+            self._provisioner = cluster_factory(provisioner=self.config.provisioner,
+                                                clusterName=None,
+                                                zone=None,  # read from instance meta-data
+                                                nodeStorage=self.config.nodeStorage,
+                                                nodeStorageOverrides=self.config.nodeStorageOverrides,
+                                                sseKey=self.config.sseKey)
             self._provisioner.setAutoscaledNodeTypes(self.config.nodeTypes)
 
     @classmethod
@@ -1913,30 +967,27 @@ class Toil(ContextManager["Toil"]):
         :return: an instance of a concrete subclass of AbstractJobStore
         """
         name, rest = cls.parseLocator(locator)
-        if name == "file":
+        if name == 'file':
             from toil.jobStores.fileJobStore import FileJobStore
-
             return FileJobStore(rest)
-        elif name == "aws":
+        elif name == 'aws':
             from toil.jobStores.aws.jobStore import AWSJobStore
-
             return AWSJobStore(rest)
-        elif name == "google":
+        elif name == 'google':
             from toil.jobStores.googleJobStore import GoogleJobStore
-
             return GoogleJobStore(rest)
         else:
             raise RuntimeError("Unknown job store implementation '%s'" % name)
 
     @staticmethod
     def parseLocator(locator: str) -> Tuple[str, str]:
-        if locator[0] in "/." or ":" not in locator:
-            return "file", locator
+        if locator[0] in '/.' or ':' not in locator:
+            return 'file', locator
         else:
             try:
-                name, rest = locator.split(":", 1)
+                name, rest = locator.split(':', 1)
             except ValueError:
-                raise RuntimeError("Invalid job store locator syntax.")
+                raise RuntimeError('Invalid job store locator syntax.')
             else:
                 return name, rest
 
@@ -1944,7 +995,7 @@ class Toil(ContextManager["Toil"]):
     def buildLocator(name: str, rest: str) -> str:
         if ":" in name:
             raise ValueError(f"Can't have a ':' in the name: '{name}'.")
-        return f"{name}:{rest}"
+        return f'{name}:{rest}'
 
     @classmethod
     def resumeJobStore(cls, locator: str) -> "AbstractJobStore":
@@ -1961,40 +1012,30 @@ class Toil(ContextManager["Toil"]):
 
         :return: an instance of a concrete subclass of AbstractBatchSystem
         """
-        kwargs = dict(
-            config=config,
-            maxCores=config.maxCores,
-            maxMemory=config.maxMemory,
-            maxDisk=config.maxDisk,
-        )
+        kwargs = dict(config=config,
+                      maxCores=config.maxCores,
+                      maxMemory=config.maxMemory,
+                      maxDisk=config.maxDisk)
 
-        from toil.batchSystems.registry import (get_batch_system,
-                                                get_batch_systems)
+        from toil.batchSystems.registry import get_batch_system, get_batch_systems
 
         try:
             batch_system = get_batch_system(config.batchSystem)
         except KeyError:
-            raise RuntimeError(
-                f"Unrecognized batch system: {config.batchSystem}  "
-                f'(choose from: {", ".join(get_batch_systems())})'
-            )
+            raise RuntimeError(f'Unrecognized batch system: {config.batchSystem}  '
+                               f'(choose from: {", ".join(get_batch_systems())})')
 
         if config.caching and not batch_system.supportsWorkerCleanup():
-            raise RuntimeError(
-                f"{config.batchSystem} currently does not support shared caching, because it "
-                "does not support cleaning up a worker after the last job finishes. Set "
-                "--caching=false"
-            )
-
-        logger.debug(
-            "Using the %s"
-            % re.sub("([a-z])([A-Z])", r"\g<1> \g<2>", batch_system.__name__).lower()
-        )
+            raise RuntimeError(f'{config.batchSystem} currently does not support shared caching, because it '
+                               'does not support cleaning up a worker after the last job finishes. Set '
+                               '--caching=false')
+
+        logger.debug('Using the %s' % re.sub("([a-z])([A-Z])", r"\g<1> \g<2>", batch_system.__name__).lower())
 
         return batch_system(**kwargs)
 
     def _setupAutoDeployment(
-        self, userScript: Optional["ModuleDescriptor"] = None
+            self, userScript: Optional["ModuleDescriptor"] = None
     ) -> None:
         """
         Determine the user script, save it to the job store and inject a reference to the saved copy into the batch system.
@@ -2007,106 +1048,86 @@ class Toil(ContextManager["Toil"]):
         if userScript is not None:
             # This branch is hit when a workflow is being started
             if userScript.belongsToToil:
-                logger.debug(
-                    "User script %s belongs to Toil. No need to auto-deploy it.",
-                    userScript,
-                )
+                logger.debug('User script %s belongs to Toil. No need to auto-deploy it.', userScript)
                 userScript = None
             else:
-                if (
-                    self._batchSystem.supportsAutoDeployment()
-                    and not self.config.disableAutoDeployment
-                ):
+                if (self._batchSystem.supportsAutoDeployment() and
+                        not self.config.disableAutoDeployment):
                     # Note that by saving the ModuleDescriptor, and not the Resource we allow for
                     # redeploying a potentially modified user script on workflow restarts.
-                    with self._jobStore.write_shared_file_stream("userScript") as f:
+                    with self._jobStore.write_shared_file_stream('userScript') as f:
                         pickle.dump(userScript, f, protocol=pickle.HIGHEST_PROTOCOL)
                 else:
                     from toil.batchSystems.singleMachine import \
                         SingleMachineBatchSystem
-
                     if not isinstance(self._batchSystem, SingleMachineBatchSystem):
-                        logger.warning(
-                            "Batch system does not support auto-deployment. The user script "
-                            "%s will have to be present at the same location on every worker.",
-                            userScript,
-                        )
+                        logger.warning('Batch system does not support auto-deployment. The user script '
+                                       '%s will have to be present at the same location on every worker.', userScript)
                     userScript = None
         else:
             # This branch is hit on restarts
-            if (
-                self._batchSystem.supportsAutoDeployment()
-                and not self.config.disableAutoDeployment
-            ):
+            if self._batchSystem.supportsAutoDeployment() and not self.config.disableAutoDeployment:
                 # We could deploy a user script
                 from toil.jobStores.abstractJobStore import NoSuchFileException
-
                 try:
-                    with self._jobStore.read_shared_file_stream("userScript") as f:
+                    with self._jobStore.read_shared_file_stream('userScript') as f:
                         userScript = safeUnpickleFromStream(f)
                 except NoSuchFileException:
-                    logger.debug(
-                        "User script neither set explicitly nor present in the job store."
-                    )
+                    logger.debug('User script neither set explicitly nor present in the job store.')
                     userScript = None
         if userScript is None:
-            logger.debug("No user script to auto-deploy.")
+            logger.debug('No user script to auto-deploy.')
         else:
-            logger.debug("Saving user script %s as a resource", userScript)
+            logger.debug('Saving user script %s as a resource', userScript)
             userScriptResource = userScript.saveAsResourceTo(self._jobStore)
-            logger.debug(
-                "Injecting user script %s into batch system.", userScriptResource
-            )
+            logger.debug('Injecting user script %s into batch system.', userScriptResource)
             self._batchSystem.setUserScript(userScriptResource)
 
     # Importing a file with a shared file name returns None, but without one it
     # returns a file ID. Explain this to MyPy.
 
     @overload
-    def importFile(
-        self, srcUrl: str, sharedFileName: str, symlink: bool = True
-    ) -> None:
+    def importFile(self,
+                   srcUrl: str,
+                   sharedFileName: str,
+                   symlink: bool = True) -> None:
         ...
 
     @overload
-    def importFile(
-        self, srcUrl: str, sharedFileName: None = None, symlink: bool = True
-    ) -> FileID:
+    def importFile(self,
+                   srcUrl: str,
+                   sharedFileName: None = None,
+                   symlink: bool = True) -> FileID:
         ...
 
-    @deprecated(new_function_name="import_file")
-    def importFile(
-        self, srcUrl: str, sharedFileName: Optional[str] = None, symlink: bool = True
-    ) -> Optional[FileID]:
+    @deprecated(new_function_name='import_file')
+    def importFile(self,
+                   srcUrl: str,
+                   sharedFileName: Optional[str] = None,
+                   symlink: bool = True) -> Optional[FileID]:
         return self.import_file(srcUrl, sharedFileName, symlink)
 
     @overload
-    def import_file(
-        self,
-        src_uri: str,
-        shared_file_name: str,
-        symlink: bool = True,
-        check_existence: bool = True,
-    ) -> None:
+    def import_file(self,
+                    src_uri: str,
+                    shared_file_name: str,
+                    symlink: bool = True,
+                    check_existence: bool = True) -> None:
         ...
 
     @overload
-    def import_file(
-        self,
-        src_uri: str,
-        shared_file_name: None = None,
-        symlink: bool = True,
-        check_existence: bool = True,
-    ) -> FileID:
+    def import_file(self,
+                    src_uri: str,
+                    shared_file_name: None = None,
+                    symlink: bool = True,
+                    check_existence: bool = True) -> FileID:
         ...
 
-    def import_file(
-        self,
-        src_uri: str,
-        shared_file_name: Optional[str] = None,
-        symlink: bool = True,
-        check_existence: bool = True,
-    ) -> Optional[FileID]:
+    def import_file(self,
+                    src_uri: str,
+                    shared_file_name: Optional[str] = None,
+                    symlink: bool = True,
+                    check_existence: bool = True) -> Optional[FileID]:
         """
         Import the file at the given URL into the job store.
 
@@ -2122,9 +1143,7 @@ class Toil(ContextManager["Toil"]):
         self._assertContextManagerUsed()
         full_uri = self.normalize_uri(src_uri, check_existence=check_existence)
         try:
-            imported = self._jobStore.import_file(
-                full_uri, shared_file_name=shared_file_name, symlink=symlink
-            )
+            imported = self._jobStore.import_file(full_uri, shared_file_name=shared_file_name, symlink=symlink)
         except FileNotFoundError:
             # TODO: I thought we refactored the different job store import
             # methods to not raise and instead return None, but that looks to
@@ -2141,10 +1160,10 @@ class Toil(ContextManager["Toil"]):
             # We need to protect the caller from missing files.
             # We think a file was missing, and we got None becasuse of it.
             # We didn't get None instead because of usign a shared file name.
-            raise FileNotFoundError(f"Could not find file {src_uri}")
+            raise FileNotFoundError(f'Could not find file {src_uri}')
         return imported
 
-    @deprecated(new_function_name="export_file")
+    @deprecated(new_function_name='export_file')
     def exportFile(self, jobStoreFileID: FileID, dstUrl: str) -> None:
         return self.export_file(jobStoreFileID, dstUrl)
 
@@ -2171,7 +1190,7 @@ class Toil(ContextManager["Toil"]):
             uri = unquote(urlparse(uri).path)  # this should strip off the local file scheme; it will be added back
 
         # account for the scheme-less case, which should be coerced to a local absolute path
-        if urlparse(uri).scheme == "":
+        if urlparse(uri).scheme == '':
             abs_path = os.path.abspath(uri)
             if not os.path.exists(abs_path) and check_existence:
                 raise FileNotFoundError(
@@ -2190,19 +1209,15 @@ class Toil(ContextManager["Toil"]):
     def _serialiseEnv(self) -> None:
         """Put the environment in a globally accessible pickle file."""
         # Dump out the environment of this process in the environment pickle file.
-        with self._jobStore.write_shared_file_stream(
-            "environment.pickle"
-        ) as fileHandle:
+        with self._jobStore.write_shared_file_stream("environment.pickle") as fileHandle:
             pickle.dump(dict(os.environ), fileHandle, pickle.HIGHEST_PROTOCOL)
         logger.debug("Written the environment for the jobs to the environment file")
 
     def _cacheAllJobs(self) -> None:
         """Download all jobs in the current job store into self.jobCache."""
-        logger.debug("Caching all jobs in job store")
-        self._jobCache = {
-            jobDesc.jobStoreID: jobDesc for jobDesc in self._jobStore.jobs()
-        }
-        logger.debug(f"{len(self._jobCache)} jobs downloaded.")
+        logger.debug('Caching all jobs in job store')
+        self._jobCache = {jobDesc.jobStoreID: jobDesc for jobDesc in self._jobStore.jobs()}
+        logger.debug(f'{len(self._jobCache)} jobs downloaded.')
 
     def _cacheJob(self, job: "JobDescription") -> None:
         """
@@ -2224,22 +1239,14 @@ class Toil(ContextManager["Toil"]):
         :param configWorkDir: Value passed to the program using the --workDir flag
         :return: Path to the Toil work directory, constant across all machines
         """
-        workDir = (
-            os.getenv("TOIL_WORKDIR_OVERRIDE")
-            or configWorkDir
-            or os.getenv("TOIL_WORKDIR")
-            or tempfile.gettempdir()
-        )
+        workDir = os.getenv('TOIL_WORKDIR_OVERRIDE') or configWorkDir or os.getenv(
+            'TOIL_WORKDIR') or tempfile.gettempdir()
         if not os.path.exists(workDir):
-            raise RuntimeError(
-                f"The directory specified by --workDir or TOIL_WORKDIR ({workDir}) does not exist."
-            )
+            raise RuntimeError(f'The directory specified by --workDir or TOIL_WORKDIR ({workDir}) does not exist.')
         return workDir
 
     @classmethod
-    def get_toil_coordination_dir(
-        cls, config_work_dir: Optional[str], config_coordination_dir: Optional[str]
-    ) -> str:
+    def get_toil_coordination_dir(cls, config_work_dir: Optional[str], config_coordination_dir: Optional[str]) -> str:
         """
         Return a path to a writable directory, which will be in memory if
         convenient. Ought to be used for file locking and coordination.
@@ -2254,54 +1261,40 @@ class Toil(ContextManager["Toil"]):
                  deleted.
         """
 
-        if "XDG_RUNTIME_DIR" in os.environ and not os.path.exists(
-            os.environ["XDG_RUNTIME_DIR"]
-        ):
+        if 'XDG_RUNTIME_DIR' in os.environ and not os.path.exists(os.environ['XDG_RUNTIME_DIR']):
             # Slurm has been observed providing this variable but not keeping
             # the directory live as long as we run for.
-            logger.warning(
-                "XDG_RUNTIME_DIR is set to nonexistent directory %s; your environment may be out of spec!",
-                os.environ["XDG_RUNTIME_DIR"],
-            )
+            logger.warning('XDG_RUNTIME_DIR is set to nonexistent directory %s; your environment may be out of spec!',
+                           os.environ['XDG_RUNTIME_DIR'])
 
         # Go get a coordination directory, using a lot of short-circuiting of
         # or and the fact that and returns its second argument when it
         # succeeds.
         coordination_dir: Optional[str] = (
             # First try an override env var
-            os.getenv("TOIL_COORDINATION_DIR_OVERRIDE")
-            or
-            # Then the value from the config
-            config_coordination_dir
-            or
-            # Then a normal env var
-            # TODO: why/how would this propagate when not using single machine?
-            os.getenv("TOIL_COORDINATION_DIR")
-            or
-            # Then try a `toil` subdirectory of the XDG runtime directory
-            # (often /var/run/users/<UID>). But only if we are actually in a
-            # session that has the env var set. Otherwise it might belong to a
-            # different set of sessions and get cleaned up out from under us
-            # when that session ends.
-            # We don't think Slurm XDG sessions are trustworthy, depending on
-            # the cluster's PAM configuration, so don't use them.
-            (
-                "XDG_RUNTIME_DIR" in os.environ
-                and "SLURM_JOBID" not in os.environ
-                and try_path(os.path.join(os.environ["XDG_RUNTIME_DIR"], "toil"))
-            )
-            or
-            # Try under /run/lock. It might be a temp dir style sticky directory.
-            try_path("/run/lock")
-            or
-            # Finally, fall back on the work dir and hope it's a legit filesystem.
-            cls.getToilWorkDir(config_work_dir)
+                os.getenv('TOIL_COORDINATION_DIR_OVERRIDE') or
+                # Then the value from the config
+                config_coordination_dir or
+                # Then a normal env var
+                # TODO: why/how would this propagate when not using single machine?
+                os.getenv('TOIL_COORDINATION_DIR') or
+                # Then try a `toil` subdirectory of the XDG runtime directory
+                # (often /var/run/users/<UID>). But only if we are actually in a
+                # session that has the env var set. Otherwise it might belong to a
+                # different set of sessions and get cleaned up out from under us
+                # when that session ends.
+                # We don't think Slurm XDG sessions are trustworthy, depending on
+                # the cluster's PAM configuration, so don't use them.
+                ('XDG_RUNTIME_DIR' in os.environ and 'SLURM_JOBID' not in os.environ and try_path(
+                    os.path.join(os.environ['XDG_RUNTIME_DIR'], 'toil'))) or
+                # Try under /run/lock. It might be a temp dir style sticky directory.
+                try_path('/run/lock') or
+                # Finally, fall back on the work dir and hope it's a legit filesystem.
+                cls.getToilWorkDir(config_work_dir)
         )
 
         if coordination_dir is None:
-            raise RuntimeError(
-                "Could not determine a coordination directory by any method!"
-            )
+            raise RuntimeError("Could not determine a coordination directory by any method!")
 
         return coordination_dir
 
@@ -2315,11 +1308,11 @@ class Toil(ContextManager["Toil"]):
 
         :param workflow_id: The ID of the current Toil workflow.
         """
-        return str(uuid.uuid5(uuid.UUID(getNodeID()), workflow_id)).replace("-", "")
+        return str(uuid.uuid5(uuid.UUID(getNodeID()), workflow_id)).replace('-', '')
 
     @classmethod
     def getLocalWorkflowDir(
-        cls, workflowID: str, configWorkDir: Optional[str] = None
+            cls, workflowID: str, configWorkDir: Optional[str] = None
     ) -> str:
         """
         Return the directory where worker directories and the cache will be located for this workflow on this machine.
@@ -2332,9 +1325,7 @@ class Toil(ContextManager["Toil"]):
 
         # Create a directory unique to each host in case workDir is on a shared FS.
         # This prevents workers on different nodes from erasing each other's directories.
-        workflowDir: str = os.path.join(
-            base, cls._get_workflow_path_component(workflowID)
-        )
+        workflowDir: str = os.path.join(base, cls._get_workflow_path_component(workflowID))
         try:
             # Directory creation is atomic
             os.mkdir(workflowDir)
@@ -2343,17 +1334,15 @@ class Toil(ContextManager["Toil"]):
                 # The directory exists if a previous worker set it up.
                 raise
         else:
-            logger.debug(
-                "Created the workflow directory for this machine at %s" % workflowDir
-            )
+            logger.debug('Created the workflow directory for this machine at %s' % workflowDir)
         return workflowDir
 
     @classmethod
     def get_local_workflow_coordination_dir(
-        cls,
-        workflow_id: str,
-        config_work_dir: Optional[str],
-        config_coordination_dir: Optional[str],
+            cls,
+            workflow_id: str,
+            config_work_dir: Optional[str],
+            config_coordination_dir: Optional[str]
     ) -> str:
         """
         Return the directory where coordination files should be located for
@@ -2393,31 +1382,24 @@ class Toil(ContextManager["Toil"]):
         """
         logProcessContext(self.config)
 
-        with RealtimeLogger(
-            self._batchSystem,
-            level=self.options.logLevel if self.options.realTimeLogging else None,
-        ):
+        with RealtimeLogger(self._batchSystem,
+                            level=self.options.logLevel if self.options.realTimeLogging else None):
             # FIXME: common should not import from leader
             from toil.leader import Leader
-
-            return Leader(
-                config=self.config,
-                batchSystem=self._batchSystem,
-                provisioner=self._provisioner,
-                jobStore=self._jobStore,
-                rootJob=rootJob,
-                jobCache=self._jobCache,
-            ).run()
+            return Leader(config=self.config,
+                          batchSystem=self._batchSystem,
+                          provisioner=self._provisioner,
+                          jobStore=self._jobStore,
+                          rootJob=rootJob,
+                          jobCache=self._jobCache).run()
 
     def _shutdownBatchSystem(self) -> None:
         """Shuts down current batch system if it has been created."""
         startTime = time.time()
-        logger.debug("Shutting down batch system ...")
+        logger.debug('Shutting down batch system ...')
         self._batchSystem.shutdown()
-        logger.debug(
-            "... finished shutting down the batch system in %s seconds."
-            % (time.time() - startTime)
-        )
+        logger.debug('... finished shutting down the batch system in %s seconds.'
+                     % (time.time() - startTime))
 
     def _assertContextManagerUsed(self) -> None:
         if not self._inContextManager:
@@ -2432,30 +1414,25 @@ class ToilRestartException(Exception):
 class ToilContextManagerException(Exception):
     def __init__(self) -> None:
         super().__init__(
-            'This method cannot be called outside the "with Toil(...)" context manager.'
-        )
+            'This method cannot be called outside the "with Toil(...)" context manager.')
 
 
 class ToilMetrics:
-    def __init__(
-        self, bus: MessageBus, provisioner: Optional["AbstractProvisioner"] = None
-    ) -> None:
+    def __init__(self, bus: MessageBus, provisioner: Optional["AbstractProvisioner"] = None) -> None:
         clusterName = "none"
         region = "us-west-2"
         if provisioner is not None:
             clusterName = str(provisioner.clusterName)
             if provisioner._zone is not None:
-                if provisioner.cloud == "aws":
+                if provisioner.cloud == 'aws':
                     # Remove AZ name
                     region = zone_to_region(provisioner._zone)
                 else:
                     region = provisioner._zone
 
-        registry = lookupEnvVar(
-            name="docker registry",
-            envName="TOIL_DOCKER_REGISTRY",
-            defaultValue=dockerRegistry,
-        )
+        registry = lookupEnvVar(name='docker registry',
+                                envName='TOIL_DOCKER_REGISTRY',
+                                defaultValue=dockerRegistry)
 
         self.mtailImage = f"{registry}/toil-mtail:{dockerTag}"
         self.grafanaImage = f"{registry}/toil-grafana:{dockerTag}"
@@ -2472,21 +1449,14 @@ class ToilMetrics:
 
         try:
             self.mtailProc: Optional[subprocess.Popen[bytes]] = subprocess.Popen(
-                [
-                    "docker",
-                    "run",
-                    "--rm",
-                    "--interactive",
-                    "--net=host",
-                    "--name",
-                    "toil_mtail",
-                    "-p",
-                    "3903:3903",
-                    self.mtailImage,
-                ],
-                stdin=subprocess.PIPE,
-                stdout=subprocess.PIPE,
-            )
+                ["docker", "run",
+                 "--rm",
+                 "--interactive",
+                 "--net=host",
+                 "--name", "toil_mtail",
+                 "-p", "3903:3903",
+                 self.mtailImage],
+                stdin=subprocess.PIPE, stdout=subprocess.PIPE)
         except subprocess.CalledProcessError:
             logger.warning("Couldn't start toil metrics server.")
             self.mtailProc = None
@@ -2499,32 +1469,20 @@ class ToilMetrics:
         if not provisioner:
             try:
                 self.nodeExporterProc = subprocess.Popen(
-                    [
-                        "docker",
-                        "run",
-                        "--rm",
-                        "--net=host",
-                        "-p",
-                        "9100:9100",
-                        "-v",
-                        "/proc:/host/proc",
-                        "-v",
-                        "/sys:/host/sys",
-                        "-v",
-                        "/:/rootfs",
-                        "quay.io/prometheus/node-exporter:v1.3.1",
-                        "-collector.procfs",
-                        "/host/proc",
-                        "-collector.sysfs",
-                        "/host/sys",
-                        "-collector.filesystem.ignored-mount-points",
-                        "^/(sys|proc|dev|host|etc)($|/)",
-                    ]
-                )
+                    ["docker", "run",
+                     "--rm",
+                     "--net=host",
+                     "-p", "9100:9100",
+                     "-v", "/proc:/host/proc",
+                     "-v", "/sys:/host/sys",
+                     "-v", "/:/rootfs",
+                     "quay.io/prometheus/node-exporter:v1.3.1",
+                     "-collector.procfs", "/host/proc",
+                     "-collector.sysfs", "/host/sys",
+                     "-collector.filesystem.ignored-mount-points",
+                     "^/(sys|proc|dev|host|etc)($|/)"])
             except subprocess.CalledProcessError:
-                logger.warning(
-                    "Couldn't start node exporter, won't get RAM and CPU usage for dashboard."
-                )
+                logger.warning("Couldn't start node exporter, won't get RAM and CPU usage for dashboard.")
             except KeyboardInterrupt:
                 if self.nodeExporterProc is not None:
                     self.nodeExporterProc.terminate()
@@ -2541,32 +1499,23 @@ class ToilMetrics:
             JobMissingMessage: self.logMissingJob,
             JobIssuedMessage: self.logIssuedJob,
             JobFailedMessage: self.logFailedJob,
-            JobCompletedMessage: self.logCompletedJob,
+            JobCompletedMessage: self.logCompletedJob
         }
         # The only way to make this inteligible to MyPy is to wrap the dict in
         # a function that can cast.
-        MessageType = TypeVar("MessageType")
+        MessageType = TypeVar('MessageType')
 
-        def get_listener(
-            message_type: Type[MessageType],
-        ) -> Callable[[MessageType], None]:
+        def get_listener(message_type: Type[MessageType]) -> Callable[[MessageType], None]:
             return cast(Callable[[MessageType], None], TARGETS[message_type])
 
         # Then set up the listeners.
-        self._listeners = [
-            bus.subscribe(message_type, get_listener(message_type))
-            for message_type in TARGETS.keys()
-        ]
+        self._listeners = [bus.subscribe(message_type, get_listener(message_type)) for message_type in TARGETS.keys()]
 
     @staticmethod
     def _containerRunning(containerName: str) -> bool:
         try:
-            result = (
-                subprocess.check_output(
-                    ["docker", "inspect", "-f", "'{{.State.Running}}'", containerName]
-                ).decode("utf-8")
-                == "true"
-            )
+            result = subprocess.check_output(["docker", "inspect", "-f",
+                                              "'{{.State.Running}}'", containerName]).decode('utf-8') == "true"
         except subprocess.CalledProcessError:
             result = False
         return result
@@ -2578,38 +1527,24 @@ class ToilMetrics:
                     subprocess.check_call(["docker", "rm", "-f", "toil_prometheus"])
                 except subprocess.CalledProcessError:
                     pass
-                subprocess.check_call(
-                    [
-                        "docker",
-                        "run",
-                        "--name",
-                        "toil_prometheus",
-                        "--net=host",
-                        "-d",
-                        "-p",
-                        "9090:9090",
-                        self.prometheusImage,
-                        clusterName,
-                        zone,
-                    ]
-                )
+                subprocess.check_call(["docker", "run",
+                                       "--name", "toil_prometheus",
+                                       "--net=host",
+                                       "-d",
+                                       "-p", "9090:9090",
+                                       self.prometheusImage,
+                                       clusterName,
+                                       zone])
 
             if not self._containerRunning("toil_grafana"):
                 try:
                     subprocess.check_call(["docker", "rm", "-f", "toil_grafana"])
                 except subprocess.CalledProcessError:
                     pass
-                subprocess.check_call(
-                    [
-                        "docker",
-                        "run",
-                        "--name",
-                        "toil_grafana",
-                        "-d",
-                        "-p=3000:3000",
-                        self.grafanaImage,
-                    ]
-                )
+                subprocess.check_call(["docker", "run",
+                                       "--name", "toil_grafana",
+                                       "-d", "-p=3000:3000",
+                                       self.grafanaImage])
         except subprocess.CalledProcessError:
             logger.warning("Could not start prometheus/grafana dashboard.")
             return
@@ -2617,17 +1552,15 @@ class ToilMetrics:
         try:
             self.add_prometheus_data_source()
         except requests.exceptions.ConnectionError:
-            logger.debug(
-                "Could not add data source to Grafana dashboard - no metrics will be displayed."
-            )
+            logger.debug("Could not add data source to Grafana dashboard - no metrics will be displayed.")
 
     @retry(errors=[requests.exceptions.ConnectionError])
     def add_prometheus_data_source(self) -> None:
         requests.post(
-            "http://localhost:3000/api/datasources",
-            auth=("admin", "admin"),
+            'http://localhost:3000/api/datasources',
+            auth=('admin', 'admin'),
             data='{"name":"DS_PROMETHEUS","type":"prometheus", "url":"http://localhost:9090", "access":"direct"}',
-            headers={"content-type": "application/json", "access": "direct"},
+            headers={'content-type': 'application/json', "access": "direct"}
         )
 
     def log(self, message: str) -> None:
@@ -2638,10 +1571,14 @@ class ToilMetrics:
     # Note: The mtail configuration (dashboard/mtail/toil.mtail) depends on these messages
     # remaining intact
 
-    def logClusterSize(self, m: ClusterSizeMessage) -> None:
+    def logClusterSize(
+            self, m: ClusterSizeMessage
+    ) -> None:
         self.log("current_size '%s' %i" % (m.instance_type, m.current_size))
 
-    def logClusterDesiredSize(self, m: ClusterDesiredSizeMessage) -> None:
+    def logClusterDesiredSize(
+            self, m: ClusterDesiredSizeMessage
+    ) -> None:
         self.log("desired_size '%s' %i" % (m.instance_type, m.desired_size))
 
     def logQueueSize(self, m: QueueSizeMessage) -> None:
@@ -2661,107 +1598,21 @@ class ToilMetrics:
 
     def shutdown(self) -> None:
         if self.mtailProc is not None:
-            logger.debug("Stopping mtail")
+            logger.debug('Stopping mtail')
             self.mtailProc.kill()
-            logger.debug("Stopped mtail")
+            logger.debug('Stopped mtail')
         if self.nodeExporterProc is not None:
-            logger.debug("Stopping node exporter")
+            logger.debug('Stopping node exporter')
             self.nodeExporterProc.kill()
-            logger.debug("Stopped node exporter")
+            logger.debug('Stopped node exporter')
         self._listeners = []
 
 
-def parseSetEnv(l: List[str]) -> Dict[str, Optional[str]]:
-    """
-    Parse a list of strings of the form "NAME=VALUE" or just "NAME" into a dictionary.
-
-    Strings of the latter from will result in dictionary entries whose value is None.
-
-    >>> parseSetEnv([])
-    {}
-    >>> parseSetEnv(['a'])
-    {'a': None}
-    >>> parseSetEnv(['a='])
-    {'a': ''}
-    >>> parseSetEnv(['a=b'])
-    {'a': 'b'}
-    >>> parseSetEnv(['a=a', 'a=b'])
-    {'a': 'b'}
-    >>> parseSetEnv(['a=b', 'c=d'])
-    {'a': 'b', 'c': 'd'}
-    >>> parseSetEnv(['a=b=c'])
-    {'a': 'b=c'}
-    >>> parseSetEnv([''])
-    Traceback (most recent call last):
-    ...
-    ValueError: Empty name
-    >>> parseSetEnv(['=1'])
-    Traceback (most recent call last):
-    ...
-    ValueError: Empty name
-    """
-    d = {}
-    v: Optional[str] = None
-    for i in l:
-        try:
-            k, v = i.split("=", 1)
-        except ValueError:
-            k, v = i, None
-        if not k:
-            raise ValueError("Empty name")
-        d[k] = v
-    return d
-
-
-def iC(minValue: int, maxValue: Optional[int] = None) -> Callable[[int], bool]:
-    """Returns a function that checks if a given int is in the given half-open interval."""
-    if not isinstance(minValue, int):
-        raise RuntimeError(
-            f"minValue must be of type 'int', was type: {type(minValue)}: {minValue!r}."
-        )
-    if maxValue is None:
-        return lambda x: minValue <= x
-    if not isinstance(maxValue, int):
-        raise RuntimeError(
-            f"maxValue must be of type 'int' (or None), was type: {type(maxValue)}: {maxValue!r}."
-        )
-    return lambda x: minValue <= x < maxValue
-
-
-def fC(minValue: float, maxValue: Optional[float] = None) -> Callable[[float], bool]:
-    """Returns a function that checks if a given float is in the given half-open interval."""
-    if not isinstance(minValue, float):
-        raise RuntimeError(
-            f"minValue must be of type 'float', was type: {type(minValue)}: {minValue!r}."
-        )
-    if maxValue is None:
-        return lambda x: minValue <= x
-    if not isinstance(maxValue, float):
-        raise RuntimeError(
-            f"maxValue must be of type 'float' (or None), was type: {type(maxValue)}: {maxValue!r}."
-        )
-    return lambda x: minValue <= x < maxValue
-
-
-def parse_accelerator_list(specs: Optional[str]) -> List["AcceleratorRequirement"]:
-    """
-    Parse a string description of one or more accelerator requirements.
-    """
-
-    if specs is None or len(specs) == 0:
-        # Not specified, so the default default is to not need any.
-        return []
-    # Otherwise parse each requirement.
-    from toil.job import parse_accelerator
-
-    return [parse_accelerator(r) for r in specs.split(",")]
-
-
 def cacheDirName(workflowID: str) -> str:
     """
     :return: Name of the cache directory.
     """
-    return f"cache-{workflowID}"
+    return f'cache-{workflowID}'
 
 
 def getDirSizeRecursively(dirPath: str) -> int:
@@ -2789,16 +1640,8 @@ def getDirSizeRecursively(dirPath: str) -> int:
     # multiply this by 512 to return the filesize in bytes.
 
     try:
-        return (
-            int(
-                subprocess.check_output(
-                    ["du", "-s", dirPath], env=dict(os.environ, BLOCKSIZE="512")
-                )
-                .decode("utf-8")
-                .split()[0]
-            )
-            * 512
-        )
+        return int(subprocess.check_output(['du', '-s', dirPath],
+                                           env=dict(os.environ, BLOCKSIZE='512')).decode('utf-8').split()[0]) * 512
     except subprocess.CalledProcessError:
         # Something was inaccessible or went away
         return 0
@@ -2812,7 +1655,7 @@ def getFileSystemSize(dirPath: str) -> Tuple[int, int]:
     :return: free space and total size of file system
     """
     if not os.path.exists(dirPath):
-        raise RuntimeError(f"Could not find dir size for non-existent path: {dirPath}")
+        raise RuntimeError(f'Could not find dir size for non-existent path: {dirPath}')
     diskStats = os.statvfs(dirPath)
     freeSpace = diskStats.f_frsize * diskStats.f_bavail
     diskSize = diskStats.f_frsize * diskStats.f_blocks
diff --git a/src/toil/cwl/cwltoil.py b/src/toil/cwl/cwltoil.py
index 0ceb0e03..7d4b5f07 100644
--- a/src/toil/cwl/cwltoil.py
+++ b/src/toil/cwl/cwltoil.py
@@ -17,7 +17,6 @@
 
 # For an overview of how this all works, see discussion in
 # docs/architecture.rst
-import argparse
 import base64
 import copy
 import datetime
@@ -69,7 +68,7 @@ import cwltool.load_tool
 import cwltool.main
 import cwltool.resolver
 import schema_salad.ref_resolver
-from configargparse import ArgParser
+from configargparse import ArgParser, SUPPRESS, Namespace
 from cwltool.loghandler import _logger as cwllogger
 from cwltool.loghandler import defaultStreamHandler
 from cwltool.mpi import MpiConfig
@@ -128,7 +127,6 @@ from toil.jobStores.utils import JobStoreUnavailableException, generate_locator
 from toil.lib.io import mkdtemp
 from toil.lib.threading import ExceptionalThread
 from toil.statsAndLogging import DEFAULT_LOGLEVEL
-from toil.version import baseVersion
 
 logger = logging.getLogger(__name__)
 
@@ -1329,7 +1327,7 @@ class ToilFsAccess(StdFsAccess):
     def open(self, fn: str, mode: str) -> IO[Any]:
         if "w" in mode or "x" in mode or "+" in mode or "a" in mode:
             raise RuntimeError(f"Mode {mode} for opening {fn} involves writing")
-        
+
         parse = urlparse(fn)
         if parse.scheme in ["", "file"]:
             # Handle local files
@@ -1663,7 +1661,7 @@ def toil_get_file(
         index[src_path] = uri
         existing[uri] = src_path
         return schema_salad.ref_resolver.file_uri(src_path)
-    
+
 def write_file(
     writeFunc: Callable[[str], FileID],
     index: Dict[str, str],
@@ -3459,335 +3457,7 @@ usage_message = "\n\n" + textwrap.dedent(
     ]
 )
 
-
-def add_cwl_options(parser: argparse.ArgumentParser) -> None:
-    parser.add_argument("cwltool", type=str, help="CWL file to run.")
-    parser.add_argument(
-        "cwljob",
-        nargs="*",
-        help="Input file or CWL options. If CWL workflow takes an input, "
-        "the name of the input can be used as an option. "
-        'For example: "%(prog)s workflow.cwl --file1 file". '
-        "If an input has the same name as a Toil option, pass '--' before it.",
-    )
-
-    parser.add_argument("--not-strict", action="store_true")
-    parser.add_argument(
-        "--enable-dev",
-        action="store_true",
-        help="Enable loading and running development versions of CWL",
-    )
-    parser.add_argument(
-        "--enable-ext",
-        action="store_true",
-        help="Enable loading and running 'cwltool:' extensions to the CWL standards.",
-        default=False,
-    )
-    parser.add_argument("--quiet", dest="quiet", action="store_true", default=False)
-    parser.add_argument("--basedir", type=str)  # TODO: Might be hard-coded?
-    parser.add_argument("--outdir", type=str, default=None)
-    parser.add_argument("--version", action="version", version=baseVersion)
-    parser.add_argument(
-        "--log-dir",
-        type=str,
-        default="",
-        help="Log your tools stdout/stderr to this location outside of container",
-    )
-    dockergroup = parser.add_mutually_exclusive_group()
-    dockergroup.add_argument(
-        "--user-space-docker-cmd",
-        help="(Linux/OS X only) Specify a user space docker command (like "
-        "udocker or dx-docker) that will be used to call 'pull' and 'run'",
-    )
-    dockergroup.add_argument(
-        "--singularity",
-        action="store_true",
-        default=False,
-        help="Use Singularity runtime for running containers. "
-        "Requires Singularity v2.6.1+ and Linux with kernel version v3.18+ or "
-        "with overlayfs support backported.",
-    )
-    dockergroup.add_argument(
-        "--podman",
-        action="store_true",
-        default=False,
-        help="Use Podman runtime for running containers. ",
-    )
-    dockergroup.add_argument(
-        "--no-container",
-        action="store_true",
-        help="Do not execute jobs in a "
-        "Docker container, even when `DockerRequirement` "
-        "is specified under `hints`.",
-    )
-    dockergroup.add_argument(
-        "--leave-container",
-        action="store_false",
-        default=True,
-        help="Do not delete Docker container used by jobs after they exit",
-        dest="rm_container",
-    )
-    extra_dockergroup = parser.add_argument_group()
-    extra_dockergroup.add_argument(
-        "--custom-net",
-        help="Specify docker network name to pass to docker run command",
-    )
-    cidgroup = parser.add_argument_group(
-        "Options for recording the Docker container identifier into a file."
-    )
-    cidgroup.add_argument(
-        # Disabled as containerid is now saved by default
-        "--record-container-id",
-        action="store_true",
-        default=False,
-        help=argparse.SUPPRESS,
-        dest="record_container_id",
-    )
-
-    cidgroup.add_argument(
-        "--cidfile-dir",
-        type=str,
-        help="Store the Docker container ID into a file in the specified directory.",
-        default=None,
-        dest="cidfile_dir",
-    )
-
-    cidgroup.add_argument(
-        "--cidfile-prefix",
-        type=str,
-        help="Specify a prefix to the container ID filename. "
-        "Final file name will be followed by a timestamp. "
-        "The default is no prefix.",
-        default=None,
-        dest="cidfile_prefix",
-    )
-
-    parser.add_argument(
-        "--preserve-environment",
-        type=str,
-        nargs="+",
-        help="Preserve specified environment variables when running"
-        " CommandLineTools",
-        metavar=("VAR1 VAR2"),
-        default=("PATH",),
-        dest="preserve_environment",
-    )
-    parser.add_argument(
-        "--preserve-entire-environment",
-        action="store_true",
-        help="Preserve all environment variable when running CommandLineTools.",
-        default=False,
-        dest="preserve_entire_environment",
-    )
-    parser.add_argument(
-        "--destBucket",
-        type=str,
-        help="Specify a cloud bucket endpoint for output files.",
-    )
-    parser.add_argument("--beta-dependency-resolvers-configuration", default=None)
-    parser.add_argument("--beta-dependencies-directory", default=None)
-    parser.add_argument("--beta-use-biocontainers", default=None, action="store_true")
-    parser.add_argument("--beta-conda-dependencies", default=None, action="store_true")
-    parser.add_argument(
-        "--tmpdir-prefix",
-        type=str,
-        help="Path prefix for temporary directories",
-        default=DEFAULT_TMPDIR_PREFIX,
-    )
-    parser.add_argument(
-        "--tmp-outdir-prefix",
-        type=str,
-        help="Path prefix for intermediate output directories",
-        default=DEFAULT_TMPDIR_PREFIX,
-    )
-    parser.add_argument(
-        "--force-docker-pull",
-        action="store_true",
-        default=False,
-        dest="force_docker_pull",
-        help="Pull latest docker image even if it is locally present",
-    )
-    parser.add_argument(
-        "--no-match-user",
-        action="store_true",
-        default=False,
-        help="Disable passing the current uid to `docker run --user`",
-    )
-    parser.add_argument(
-        "--no-read-only",
-        action="store_true",
-        default=False,
-        help="Do not set root directory in the container as read-only",
-    )
-    parser.add_argument(
-        "--strict-memory-limit",
-        action="store_true",
-        help="When running with "
-        "software containers and the Docker engine, pass either the "
-        "calculated memory allocation from ResourceRequirements or the "
-        "default of 1 gigabyte to Docker's --memory option.",
-    )
-    parser.add_argument(
-        "--strict-cpu-limit",
-        action="store_true",
-        help="When running with "
-        "software containers and the Docker engine, pass either the "
-        "calculated cpu allocation from ResourceRequirements or the "
-        "default of 1 core to Docker's --cpu option. "
-        "Requires docker version >= v1.13.",
-    )
-    parser.add_argument(
-        "--relax-path-checks",
-        action="store_true",
-        default=False,
-        help="Relax requirements on path names to permit "
-        "spaces and hash characters.",
-        dest="relax_path_checks",
-    )
-    parser.add_argument(
-        "--default-container",
-        help="Specify a default docker container that will be "
-        "used if the workflow fails to specify one.",
-    )
-    parser.add_argument(
-        "--disable-validate",
-        dest="do_validate",
-        action="store_false",
-        default=True,
-        help=argparse.SUPPRESS,
-    )
-    parser.add_argument(
-        "--fast-parser",
-        dest="fast_parser",
-        action="store_true",
-        default=False,
-        help=argparse.SUPPRESS,
-    )
-    checkgroup = parser.add_mutually_exclusive_group()
-    checkgroup.add_argument(
-        "--compute-checksum",
-        action="store_true",
-        default=True,
-        help="Compute checksum of contents while collecting outputs",
-        dest="compute_checksum",
-    )
-    checkgroup.add_argument(
-        "--no-compute-checksum",
-        action="store_false",
-        help="Do not compute checksum of contents while collecting outputs",
-        dest="compute_checksum",
-    )
-
-    parser.add_argument(
-        "--eval-timeout",
-        help="Time to wait for a Javascript expression to evaluate before giving "
-        "an error, default 20s.",
-        type=float,
-        default=20,
-    )
-    parser.add_argument(
-        "--overrides",
-        type=str,
-        default=None,
-        help="Read process requirement overrides from file.",
-    )
-
-    parser.add_argument(
-        "--mpi-config-file",
-        type=str,
-        default=None,
-        help="Platform specific configuration for MPI (parallel "
-        "launcher, its flag etc). See the cwltool README "
-        "section 'Running MPI-based tools' for details of the format: "
-        "https://github.com/common-workflow-language/cwltool#running-mpi-based-tools-that-need-to-be-launched",
-    )
-    parser.add_argument(
-        "--bypass-file-store",
-        action="store_true",
-        default=False,
-        help="Do not use Toil's file store and assume all "
-        "paths are accessible in place from all nodes.",
-        dest="bypass_file_store",
-    )
-    parser.add_argument(
-        "--reference-inputs",
-        action="store_true",
-        default=False,
-        help="Do not copy remote inputs into Toil's file "
-        "store and assume they are accessible in place from "
-        "all nodes.",
-        dest="reference_inputs",
-    )
-    parser.add_argument(
-        "--disable-streaming",
-        action="store_true",
-        default=False,
-        help="Disable file streaming for files that have 'streamable' flag True",
-        dest="disable_streaming",
-    )
-
-    provgroup = parser.add_argument_group(
-        "Options for recording provenance information of the execution"
-    )
-    provgroup.add_argument(
-        "--provenance",
-        help="Save provenance to specified folder as a "
-        "Research Object that captures and aggregates "
-        "workflow execution and data products.",
-        type=str,
-    )
-
-    provgroup.add_argument(
-        "--enable-user-provenance",
-        default=False,
-        action="store_true",
-        help="Record user account info as part of provenance.",
-        dest="user_provenance",
-    )
-    provgroup.add_argument(
-        "--disable-user-provenance",
-        default=False,
-        action="store_false",
-        help="Do not record user account info in provenance.",
-        dest="user_provenance",
-    )
-    provgroup.add_argument(
-        "--enable-host-provenance",
-        default=False,
-        action="store_true",
-        help="Record host info as part of provenance.",
-        dest="host_provenance",
-    )
-    provgroup.add_argument(
-        "--disable-host-provenance",
-        default=False,
-        action="store_false",
-        help="Do not record host info in provenance.",
-        dest="host_provenance",
-    )
-    provgroup.add_argument(
-        "--orcid",
-        help="Record user ORCID identifier as part of "
-        "provenance, e.g. https://orcid.org/0000-0002-1825-0097 "
-        "or 0000-0002-1825-0097. Alternatively the environment variable "
-        "ORCID may be set.",
-        dest="orcid",
-        default=os.environ.get("ORCID", ""),
-        type=str,
-    )
-    provgroup.add_argument(
-        "--full-name",
-        help="Record full name of user as part of provenance, "
-        "e.g. Josiah Carberry. You may need to use shell quotes to preserve "
-        "spaces. Alternatively the environment variable CWL_FULL_NAME may "
-        "be set.",
-        dest="cwl_full_name",
-        default=os.environ.get("CWL_FULL_NAME", ""),
-        type=str,
-    )
-
-
-def get_options(args: List[str]) -> argparse.Namespace:
+def get_options(args: List[str]) -> Namespace:
     """
     Parse given args and properly add non-Toil arguments into the cwljob of the Namespace.
     :param args: List of args from command line
@@ -3795,9 +3465,7 @@ def get_options(args: List[str]) -> argparse.Namespace:
     """
     parser = ArgParser()
     addOptions(parser, jobstore_as_flag=True, cwl=True)
-    add_cwl_options(parser)
-
-    options: argparse.Namespace
+    options: Namespace
     options, cwl_options = parser.parse_known_args(args)
     options.cwljob.extend(cwl_options)
 
@@ -3816,12 +3484,13 @@ def main(args: Optional[List[str]] = None, stdout: TextIO = sys.stdout) -> int:
 
     # Do cwltool setup
     cwltool.main.setup_schema(args=options, custom_schema_callback=None)
+    tmpdir_prefix = options.tmpdir_prefix = options.tmpdir_prefix or DEFAULT_TMPDIR_PREFIX
 
     # We need a workdir for the CWL runtime contexts.
-    if options.tmpdir_prefix != DEFAULT_TMPDIR_PREFIX:
+    if tmpdir_prefix != DEFAULT_TMPDIR_PREFIX:
         # if tmpdir_prefix is not the default value, move
         # workdir and the default job store under it
-        workdir = cwltool.utils.create_tmp_dir(options.tmpdir_prefix)
+        workdir = cwltool.utils.create_tmp_dir(tmpdir_prefix)
     else:
         # Use a directory in the default tmpdir
         workdir = mkdtemp()
@@ -3841,13 +3510,13 @@ def main(args: Optional[List[str]] = None, stdout: TextIO = sys.stdout) -> int:
     options.do_validate = True
     options.pack = False
     options.print_subgraph = False
-    if options.tmpdir_prefix != DEFAULT_TMPDIR_PREFIX and options.workDir is None:
+    if tmpdir_prefix != DEFAULT_TMPDIR_PREFIX and options.workDir is None:
         # We need to override workDir because by default Toil will pick
         # somewhere under the system temp directory if unset, ignoring
         # --tmpdir-prefix.
         #
         # If set, workDir needs to exist, so we directly use the prefix
-        options.workDir = cwltool.utils.create_tmp_dir(options.tmpdir_prefix)
+        options.workDir = cwltool.utils.create_tmp_dir(tmpdir_prefix)
 
     if options.batchSystem == "kubernetes":
         # Containers under Kubernetes can only run in Singularity
@@ -3864,12 +3533,10 @@ def main(args: Optional[List[str]] = None, stdout: TextIO = sys.stdout) -> int:
 
     logger.debug(f"Final job store {options.jobStore} and workDir {options.workDir}")
 
-    outdir = (
-        os.path.abspath(os.getcwd())
-        if options.outdir is None
-        else os.path.abspath(options.outdir)
+    outdir = os.path.abspath(options.outdir or os.getcwd())
+    tmp_outdir_prefix = os.path.abspath(
+        options.tmp_outdir_prefix or DEFAULT_TMPDIR_PREFIX
     )
-    tmp_outdir_prefix = os.path.abspath(options.tmp_outdir_prefix)
 
     fileindex: Dict[str, str] = {}
     existing: Dict[str, str] = {}
@@ -3914,7 +3581,7 @@ def main(args: Optional[List[str]] = None, stdout: TextIO = sys.stdout) -> int:
 
     if options.provenance:
         research_obj = cwltool.cwlprov.ro.ResearchObject(
-            temp_prefix_ro=options.tmp_outdir_prefix,
+            temp_prefix_ro=tmp_outdir_prefix,
             orcid=options.orcid,
             full_name=options.cwl_full_name,
             fsaccess=runtime_context.make_fs_access(""),
@@ -4218,7 +3885,7 @@ def main(args: Optional[List[str]] = None, stdout: TextIO = sys.stdout) -> int:
 
 
 def find_default_container(
-    args: argparse.Namespace, builder: cwltool.builder.Builder
+    args: Namespace, builder: cwltool.builder.Builder
 ) -> Optional[str]:
     """Find the default constructor by consulting a Toil.options object."""
     if args.default_container:
diff --git a/src/toil/options/common.py b/src/toil/options/common.py
new file mode 100644
index 00000000..47d0ab76
--- /dev/null
+++ b/src/toil/options/common.py
@@ -0,0 +1,738 @@
+import os
+from argparse import ArgumentParser, Action, _AppendAction
+from typing import Any, Optional, Union, Type, Callable, List, Dict, TYPE_CHECKING
+
+from distutils.util import strtobool
+from configargparse import SUPPRESS
+import logging
+
+from ruamel.yaml import YAML
+
+from toil.lib.conversions import bytes2human, human2bytes
+
+from toil.batchSystems.options import add_all_batchsystem_options
+from toil.provisioners import parse_node_types
+from toil.statsAndLogging import add_logging_options
+if TYPE_CHECKING:
+    from toil.job import AcceleratorRequirement
+
+logger = logging.getLogger(__name__)
+
+# aim to pack autoscaling jobs within a 30 minute block before provisioning a new node
+defaultTargetTime = 1800
+SYS_MAX_SIZE = 9223372036854775807
+# sys.max_size on 64 bit systems is 9223372036854775807, so that 32-bit systems
+# use the same number
+
+def parse_set_env(l: List[str]) -> Dict[str, Optional[str]]:
+    """
+    Parse a list of strings of the form "NAME=VALUE" or just "NAME" into a dictionary.
+
+    Strings of the latter from will result in dictionary entries whose value is None.
+
+    >>> parse_set_env([])
+    {}
+    >>> parse_set_env(['a'])
+    {'a': None}
+    >>> parse_set_env(['a='])
+    {'a': ''}
+    >>> parse_set_env(['a=b'])
+    {'a': 'b'}
+    >>> parse_set_env(['a=a', 'a=b'])
+    {'a': 'b'}
+    >>> parse_set_env(['a=b', 'c=d'])
+    {'a': 'b', 'c': 'd'}
+    >>> parse_set_env(['a=b=c'])
+    {'a': 'b=c'}
+    >>> parse_set_env([''])
+    Traceback (most recent call last):
+    ...
+    ValueError: Empty name
+    >>> parse_set_env(['=1'])
+    Traceback (most recent call last):
+    ...
+    ValueError: Empty name
+    """
+    d = {}
+    v: Optional[str] = None
+    for i in l:
+        try:
+            k, v = i.split('=', 1)
+        except ValueError:
+            k, v = i, None
+        if not k:
+            raise ValueError('Empty name')
+        d[k] = v
+    return d
+
+
+def parse_str_list(s: str) -> List[str]:
+    return [str(x) for x in s.split(",")]
+
+
+def parse_int_list(s: str) -> List[int]:
+    return [int(x) for x in s.split(",")]
+
+
+def iC(min_value: int, max_value: Optional[int] = None) -> Callable[[int], bool]:
+    """Returns a function that checks if a given int is in the given half-open interval."""
+    assert isinstance(min_value, int)
+    if max_value is None:
+        return lambda x: min_value <= x
+    assert isinstance(max_value, int)
+    return lambda x: min_value <= x < max_value
+
+
+def fC(minValue: float, maxValue: Optional[float] = None) -> Callable[[float], bool]:
+    """Returns a function that checks if a given float is in the given half-open interval."""
+    assert isinstance(minValue, float)
+    if maxValue is None:
+        return lambda x: minValue <= x
+    assert isinstance(maxValue, float)
+    return lambda x: minValue <= x < maxValue
+
+
+def parse_accelerator_list(specs: Optional[str]) -> List['AcceleratorRequirement']:
+    """
+    Parse a string description of one or more accelerator requirements.
+    """
+
+    if specs is None or len(specs) == 0:
+        # Not specified, so the default default is to not need any.
+        return []
+    # Otherwise parse each requirement.
+    from toil.job import parse_accelerator
+
+    return [parse_accelerator(r) for r in specs.split(',')]
+
+
+def parseBool(val: str) -> bool:
+    if val.lower() in ['true', 't', 'yes', 'y', 'on', '1']:
+        return True
+    elif val.lower() in ['false', 'f', 'no', 'n', 'off', '0']:
+        return False
+    else:
+        raise RuntimeError("Could not interpret \"%s\" as a boolean value" % val)
+
+
+# This is kept in the outer scope as multiple batchsystem files use this
+def make_open_interval_action(min: Union[int, float], max: Optional[Union[int, float]] = None) -> Type[Action]:
+    """
+    Returns an argparse action class to check if the input is within the given half-open interval.
+    ex:
+    Provided value to argparse must be within the interval [min, max)
+    Types of min and max must be the same (max may be None)
+
+    :param min: float/int
+    :param max: optional float/int
+    :return: argparse action class
+    """
+
+    class IntOrFloatOpenAction(Action):
+        def __call__(self, parser: Any, namespace: Any, values: Any, option_string: Any = None) -> None:
+            if isinstance(min, int):
+                if max is not None:  # for mypy
+                    assert isinstance(max, int)
+                func = iC(min, max)
+            else:
+                func = fC(min, max)
+            try:
+                if not func(values):
+                    raise parser.error(
+                        f"{option_string} ({values}) must be within the range: [{min}, {'infinity' if max is None else max})")
+            except AssertionError:
+                raise RuntimeError(f"The {option_string} option has an invalid value: {values}")
+            setattr(namespace, self.dest, values)
+
+    return IntOrFloatOpenAction
+
+
+def parse_jobstore(jobstore_uri: str) -> str:
+    """
+    Turn the jobstore string into it's corresponding URI
+    ex:
+    /path/to/jobstore -> file:/path/to/jobstore
+
+    If the jobstore string already is a URI, return the jobstore:
+    aws:/path/to/jobstore -> aws:/path/to/jobstore
+    :param jobstore_uri: string of the jobstore
+    :return: URI of the jobstore
+    """
+    from toil.common import Toil
+    name, rest = Toil.parseLocator(jobstore_uri)
+    if name == 'file':
+        # We need to resolve relative paths early, on the leader, because the worker process
+        # may have a different working directory than the leader, e.g. under Mesos.
+        return Toil.buildLocator(name, os.path.abspath(rest))
+    else:
+        return jobstore_uri
+
+
+JOBSTORE_HELP = ("The location of the job store for the workflow.  "
+                 "A job store holds persistent information about the jobs, stats, and files in a "
+                 "workflow. If the workflow is run with a distributed batch system, the job "
+                 "store must be accessible by all worker nodes. Depending on the desired "
+                 "job store implementation, the location should be formatted according to "
+                 "one of the following schemes:\n\n"
+                 "file:<path> where <path> points to a directory on the file systen\n\n"
+                 "aws:<region>:<prefix> where <region> is the name of an AWS region like "
+                 "us-west-2 and <prefix> will be prepended to the names of any top-level "
+                 "AWS resources in use by job store, e.g. S3 buckets.\n\n "
+                 "google:<project_id>:<prefix> TODO: explain\n\n"
+                 "For backwards compatibility, you may also specify ./foo (equivalent to "
+                 "file:./foo or just file:foo) or /bar (equivalent to file:/bar).")
+
+
+def add_base_toil_options(parser: ArgumentParser, jobstore_as_flag: bool = False, cwl: bool = False) -> None:
+    """
+    Add base Toil command line options to the parser.
+    :param parser: Argument parser to add options to
+    :param jobstore_as_flag: make the job store option a --jobStore flag instead of a required jobStore positional argument.
+    :param cwl: whether CWL should be included or not
+    """
+
+    # This is necessary as the config file must have at least one valid key to parse properly and we want to use a
+    # dummy key
+    config = parser.add_argument_group()
+    config.add_argument("--config_version", default=None, help=SUPPRESS)
+
+    # If using argparse instead of configargparse, this should just not parse when calling parse_args()
+    # default config value is set to none as defaults should already be populated at config init
+    config.add_argument('--config', dest='config', is_config_file_arg=True, default=None, metavar="PATH",
+                        help="Get options from a config file.")
+
+    def convert_bool(b: str) -> bool:
+        """Convert a string representation of bool to bool"""
+        return bool(strtobool(b))
+
+    def opt_strtobool(b: Optional[str]) -> Optional[bool]:
+        """Convert an optional string representation of bool to None or bool"""
+        return b if b is None else convert_bool(b)
+
+    add_logging_options(parser)
+    parser.register("type", "bool", parseBool)  # Custom type for arg=True/False.
+
+    # Core options
+    core_options = parser.add_argument_group(
+        title="Toil core options.",
+        description="Options to specify the location of the Toil workflow and "
+                    "turn on stats collation about the performance of jobs."
+    )
+    if jobstore_as_flag:
+        core_options.add_argument('--jobstore', '--jobStore', dest='jobStore', type=parse_jobstore, default=None,
+                                  help=JOBSTORE_HELP)
+    else:
+        core_options.add_argument('jobStore', type=parse_jobstore, help=JOBSTORE_HELP)
+
+    class WorkDirAction(Action):
+        """
+        Argparse action class to check that the provided --workDir exists
+        """
+
+        def __call__(self, parser: Any, namespace: Any, values: Any, option_string: Any = None) -> None:
+            workDir = values
+            if workDir is not None:
+                workDir = os.path.abspath(workDir)
+                if not os.path.exists(workDir):
+                    raise RuntimeError(f"The path provided to --workDir ({workDir}) does not exist.")
+
+                if len(workDir) > 80:
+                    logger.warning(f'Length of workDir path "{workDir}" is {len(workDir)} characters.  '
+                                   f'Consider setting a shorter path with --workPath or setting TMPDIR to something '
+                                   f'like "/tmp" to avoid overly long paths.')
+            setattr(namespace, self.dest, workDir)
+
+    class CoordinationDirAction(Action):
+        """
+        Argparse action class to check that the provided --coordinationDir exists
+        """
+
+        def __call__(self, parser: Any, namespace: Any, values: Any, option_string: Any = None) -> None:
+            coordination_dir = values
+            if coordination_dir is not None:
+                coordination_dir = os.path.abspath(coordination_dir)
+                if not os.path.exists(coordination_dir):
+                    raise RuntimeError(
+                        f"The path provided to --coordinationDir ({coordination_dir}) does not exist.")
+            setattr(namespace, self.dest, coordination_dir)
+
+    def make_closed_interval_action(min: Union[int, float], max: Optional[Union[int, float]] = None) -> Type[Action]:
+        """
+        Returns an argparse action class to check if the input is within the given half-open interval.
+        ex:
+        Provided value to argparse must be within the interval [min, max]
+
+        :param min: int/float
+        :param max: optional int/float
+        :return: argparse action
+        """
+
+        class ClosedIntOrFloatAction(Action):
+            def __call__(self, parser: Any, namespace: Any, values: Any, option_string: Any = None) -> None:
+                def is_within(x: Union[int, float]) -> bool:
+                    if max is None:
+                        return min <= x
+                    else:
+                        return min <= x <= max
+
+                try:
+                    if not is_within(values):
+                        raise parser.error(
+                            f"{option_string} ({values}) must be within the range: [{min}, {'infinity' if max is None else max}]")
+                except AssertionError:
+                    raise RuntimeError(f"The {option_string} option has an invalid value: {values}")
+                setattr(namespace, self.dest, values)
+
+        return ClosedIntOrFloatAction
+
+    core_options.add_argument("--workDir", dest="workDir", default=None, env_var="TOIL_WORKDIR", action=WorkDirAction,
+                              metavar="PATH",
+                              help="Absolute path to directory where temporary files generated during the Toil "
+                                   "run should be placed. Standard output and error from batch system jobs "
+                                   "(unless --noStdOutErr is set) will be placed in this directory. A cache directory "
+                                   "may be placed in this directory. Temp files and folders will be placed in a "
+                                   "directory toil-<workflowID> within workDir. The workflowID is generated by "
+                                   "Toil and will be reported in the workflow logs. Default is determined by the "
+                                   "variables (TMPDIR, TEMP, TMP) via mkdtemp. This directory needs to exist on "
+                                   "all machines running jobs; if capturing standard output and error from batch "
+                                   "system jobs is desired, it will generally need to be on a shared file system. "
+                                   "When sharing a cache between containers on a host, this directory must be "
+                                   "shared between the containers.")
+    core_options.add_argument("--coordinationDir", dest="coordination_dir", default=None,
+                              env_var="TOIL_COORDINATION_DIR", action=CoordinationDirAction, metavar="PATH",
+                              help="Absolute path to directory where Toil will keep state and lock files."
+                                   "When sharing a cache between containers on a host, this directory must be "
+                                   "shared between the containers.")
+    core_options.add_argument("--noStdOutErr", dest="noStdOutErr", default=False, action="store_true",
+                              help="Do not capture standard output and error from batch system jobs.")
+    core_options.add_argument("--stats", dest="stats", default=False, action="store_true",
+                              help="Records statistics about the toil workflow to be used by 'toil stats'.")
+    clean_choices = ['always', 'onError', 'never', 'onSuccess']
+    core_options.add_argument("--clean", dest="clean", choices=clean_choices, default="onSuccess",
+                              help=f"Determines the deletion of the jobStore upon completion of the program.  "
+                                   f"Choices: {clean_choices}.  The --stats option requires information from the "
+                                   f"jobStore upon completion so the jobStore will never be deleted with that flag.  "
+                                   f"If you wish to be able to restart the run, choose \'never\' or \'onSuccess\'.  "
+                                   f"Default is \'never\' if stats is enabled, and \'onSuccess\' otherwise.")
+    core_options.add_argument("--cleanWorkDir", dest="cleanWorkDir", choices=clean_choices, default='always',
+                              help=f"Determines deletion of temporary worker directory upon completion of a job.  "
+                                   f"Choices: {clean_choices}.  Default = always.  WARNING: This option should be "
+                                   f"changed for debugging only.  Running a full pipeline with this option could "
+                                   f"fill your disk with excessive intermediate data.")
+    core_options.add_argument("--clusterStats", dest="clusterStats", nargs='?', action='store', default=None,
+                              metavar="OPT_PATH", const=os.getcwd(),
+                              help="If enabled, writes out JSON resource usage statistics to a file.  "
+                                   "The default location for this file is the current working directory, but an "
+                                   "absolute path can also be passed to specify where this file should be written. "
+                                   "This options only applies when using scalable batch systems.")
+
+    # Restarting the workflow options
+    restart_options = parser.add_argument_group(
+        title="Toil options for restarting an existing workflow.",
+        description="Allows the restart of an existing workflow"
+    )
+    restart_options.add_argument("--restart", dest="restart", default=False, action="store_true",
+                                 help="If --restart is specified then will attempt to restart existing workflow "
+                                      "at the location pointed to by the --jobStore option. Will raise an exception "
+                                      "if the workflow does not exist")
+
+    # Batch system options
+    batchsystem_options = parser.add_argument_group(
+        title="Toil options for specifying the batch system.",
+        description="Allows the specification of the batch system."
+    )
+    add_all_batchsystem_options(batchsystem_options)
+
+    # File store options
+    file_store_options = parser.add_argument_group(
+        title="Toil options for configuring storage.",
+        description="Allows configuring Toil's data storage."
+    )
+    link_imports = file_store_options.add_mutually_exclusive_group()
+    link_imports_help = ("When using a filesystem based job store, CWL input files are by default symlinked in.  "
+                         "Setting this option to True instead copies the files into the job store, which may protect "
+                         "them from being modified externally.  When set to False, as long as caching is enabled, "
+                         "Toil will protect the file automatically by changing the permissions to read-only."
+                         "default=%(default)s")
+    link_imports.add_argument("--symlinkImports", dest="symlinkImports", type=convert_bool, default=True,
+                              metavar="BOOL", help=link_imports_help)
+    move_exports = file_store_options.add_mutually_exclusive_group()
+    move_exports_help = ('When using a filesystem based job store, output files are by default moved to the '
+                         'output directory, and a symlink to the moved exported file is created at the initial '
+                         'location.  Setting this option to True instead copies the files into the output directory.  '
+                         'Applies to filesystem-based job stores only.'
+                         'default=%(default)s')
+    move_exports.add_argument("--moveOutputs", dest="moveOutputs", type=convert_bool, default=False, metavar="BOOL",
+                              help=move_exports_help)
+
+    caching = file_store_options.add_mutually_exclusive_group()
+    caching_help = "Enable or disable caching for your workflow, specifying this overrides default from job store"
+    caching.add_argument('--caching', dest='caching', type=opt_strtobool, default=None, metavar="BOOL",
+                         help=caching_help)
+    # default is None according to PR 4299, seems to be generated at runtime
+
+    # Auto scaling options
+    autoscaling_options = parser.add_argument_group(
+        title="Toil options for autoscaling the cluster of worker nodes.",
+        description="Allows the specification of the minimum and maximum number of nodes in an autoscaled cluster, "
+                    "as well as parameters to control the level of provisioning."
+    )
+    provisioner_choices = ['aws', 'gce', None]
+
+    # TODO: Better consolidate this provisioner arg and the one in provisioners/__init__.py?
+    autoscaling_options.add_argument('--provisioner', '-p', dest="provisioner", choices=provisioner_choices,
+                                     default=None,
+                                     help=f"The provisioner for cluster auto-scaling.  This is the main Toil "
+                                          f"'--provisioner' option, and defaults to None for running on single "
+                                          f"machine and non-auto-scaling batch systems.  The currently supported "
+                                          f"choices are {provisioner_choices}.  The default is %(default)s.")
+    autoscaling_options.add_argument('--nodeTypes', default=[], dest="nodeTypes", type=parse_node_types,
+                                     action="extend",
+                                     help="Specifies a list of comma-separated node types, each of which is "
+                                          "composed of slash-separated instance types, and an optional spot "
+                                          "bid set off by a colon, making the node type preemptible. Instance "
+                                          "types may appear in multiple node types, and the same node type "
+                                          "may appear as both preemptible and non-preemptible.\n"
+                                          "Valid argument specifying two node types:\n"
+                                          "\tc5.4xlarge/c5a.4xlarge:0.42,t2.large\n"
+                                          "Node types:\n"
+                                          "\tc5.4xlarge/c5a.4xlarge:0.42 and t2.large\n"
+                                          "Instance types:\n"
+                                          "\tc5.4xlarge, c5a.4xlarge, and t2.large\n"
+                                          "Semantics:\n"
+                                          "\tBid $0.42/hour for either c5.4xlarge or c5a.4xlarge instances,\n"
+                                          "\ttreated interchangeably, while they are available at that price,\n"
+                                          "\tand buy t2.large instances at full price.\n"
+                                          "default=%(default)s")
+
+    class NodeExtendAction(_AppendAction):
+        """
+        argparse Action class to remove the default value on first call, and act as an extend action after
+        """
+
+        # with action=append/extend, the argparse default is always prepended to the option
+        # so make the CLI have priority by rewriting the option on the first run
+        def __init__(self, option_strings: Any, dest: Any, **kwargs: Any):
+            super().__init__(option_strings, dest, **kwargs)
+            self.is_default = True
+
+        def __call__(self, parser: Any, namespace: Any, values: Any, option_string: Any = None) -> None:
+            if self.is_default:
+                setattr(namespace, self.dest, values)
+                self.is_default = False
+            else:
+                super().__call__(parser, namespace, values, option_string)
+
+    autoscaling_options.add_argument('--maxNodes', default=[10], dest="maxNodes", type=parse_int_list,
+                                     action=NodeExtendAction, metavar="INT[,INT...]",
+                                     help=f"Maximum number of nodes of each type in the cluster, if using autoscaling, "
+                                          f"provided as a comma-separated list.  The first value is used as a default "
+                                          f"if the list length is less than the number of nodeTypes.  "
+                                          f"default=%(default)s")
+    autoscaling_options.add_argument('--minNodes', default=[0], dest="minNodes", type=parse_int_list,
+                                     action=NodeExtendAction, metavar="INT[,INT...]",
+                                     help="Mininum number of nodes of each type in the cluster, if using "
+                                          "auto-scaling.  This should be provided as a comma-separated list of the "
+                                          "same length as the list of node types. default=%(default)s")
+    autoscaling_options.add_argument("--targetTime", dest="targetTime", default=defaultTargetTime, type=int,
+                                     action=make_closed_interval_action(0), metavar="INT",
+                                     help=f"Sets how rapidly you aim to complete jobs in seconds. Shorter times mean "
+                                          f"more aggressive parallelization. The autoscaler attempts to scale up/down "
+                                          f"so that it expects all queued jobs will complete within targetTime "
+                                          f"seconds.  default=%(default)s")
+    autoscaling_options.add_argument("--betaInertia", dest="betaInertia", default=0.1, type=float,
+                                     action=make_closed_interval_action(0.0, 0.9), metavar="FLOAT",
+                                     help=f"A smoothing parameter to prevent unnecessary oscillations in the number "
+                                          f"of provisioned nodes. This controls an exponentially weighted moving "
+                                          f"average of the estimated number of nodes. A value of 0.0 disables any "
+                                          f"smoothing, and a value of 0.9 will smooth so much that few changes will "
+                                          f"ever be made.  Must be between 0.0 and 0.9. default=%(default)s")
+    autoscaling_options.add_argument("--scaleInterval", dest="scaleInterval", default=60, type=int, metavar="INT",
+                                     help=f"The interval (seconds) between assessing if the scale of "
+                                          f"the cluster needs to change. default=%(default)s")
+    autoscaling_options.add_argument("--preemptibleCompensation", "--preemptableCompensation",
+                                     dest="preemptibleCompensation", default=0.0, type=float,
+                                     action=make_closed_interval_action(0.0, 1.0), metavar="FLOAT",
+                                     help=f"The preference of the autoscaler to replace preemptible nodes with "
+                                          f"non-preemptible nodes, when preemptible nodes cannot be started for some "
+                                          f"reason. This value must be between 0.0 and 1.0, inclusive.  "
+                                          f"A value of 0.0 disables such "
+                                          f"compensation, a value of 0.5 compensates two missing preemptible nodes "
+                                          f"with a non-preemptible one. A value of 1.0 replaces every missing "
+                                          f"pre-emptable node with a non-preemptible one. default=%(default)s")
+    autoscaling_options.add_argument("--nodeStorage", dest="nodeStorage", default=50, type=int, metavar="INT",
+                                     help="Specify the size of the root volume of worker nodes when they are launched "
+                                          "in gigabytes. You may want to set this if your jobs require a lot of disk "
+                                          f"space.  (default=%(default)s).")
+    autoscaling_options.add_argument('--nodeStorageOverrides', dest="nodeStorageOverrides", default=[],
+                                     type=parse_str_list, action="extend",
+                                     metavar="NODETYPE:NODESTORAGE[,NODETYPE:NODESTORAGE...]",
+                                     help="Comma-separated list of nodeType:nodeStorage that are used to override "
+                                          "the default value from --nodeStorage for the specified nodeType(s).  "
+                                          "This is useful for heterogeneous jobs where some tasks require much more "
+                                          "disk than others.")
+
+    autoscaling_options.add_argument("--metrics", dest="metrics", default=False, type=convert_bool, metavar="BOOL",
+                                     help="Enable the prometheus/grafana dashboard for monitoring CPU/RAM usage, "
+                                          "queue size, and issued jobs.")
+    autoscaling_options.add_argument("--assumeZeroOverhead", dest="assume_zero_overhead", default=False,
+                                     type=convert_bool, metavar="BOOL",
+                                     help="Ignore scheduler and OS overhead and assume jobs can use every last byte "
+                                          "of memory and disk on a node when autoscaling.")
+
+    # Parameters to limit service jobs / detect service deadlocks
+    service_options = parser.add_argument_group(
+        title="Toil options for limiting the number of service jobs and detecting service deadlocks",
+        description="Allows the specification of the maximum number of service jobs in a cluster.  By keeping "
+                    "this limited we can avoid nodes occupied with services causing deadlocks."
+    )
+    service_options.add_argument("--maxServiceJobs", dest="maxServiceJobs", default=SYS_MAX_SIZE, type=int,
+                                 metavar="INT",
+                                 help=SUPPRESS if cwl else f"The maximum number of service jobs that can be run "
+                                                           f"concurrently, excluding service jobs running on "
+                                                           f"preemptible nodes.  default=%(default)s")
+    service_options.add_argument("--maxPreemptibleServiceJobs", dest="maxPreemptibleServiceJobs",
+                                 default=SYS_MAX_SIZE,
+                                 type=int, metavar="INT",
+                                 help=SUPPRESS if cwl else "The maximum number of service jobs that can run "
+                                                           "concurrently on preemptible nodes.  default=%(default)s")
+    service_options.add_argument("--deadlockWait", dest="deadlockWait", default=60, type=int, metavar="INT",
+                                 help=SUPPRESS if cwl else f"Time, in seconds, to tolerate the workflow running only "
+                                                           f"the same service jobs, with no jobs to use them, "
+                                                           f"before declaring the workflow to be deadlocked and "
+                                                           f"stopping.  default=%(default)s")
+    service_options.add_argument("--deadlockCheckInterval", dest="deadlockCheckInterval", default=30, type=int,
+                                 metavar="INT",
+                                 help=SUPPRESS if cwl else "Time, in seconds, to wait between checks to see if the "
+                                                           "workflow is stuck running only service jobs, with no jobs "
+                                                           "to use them. Should be shorter than --deadlockWait. May "
+                                                           "need to be increased if the batch system cannot enumerate "
+                                                           "running jobs quickly enough, or if polling for running "
+                                                           "jobs is placing an unacceptable load on a shared cluster."
+                                                           f"default=%(default)s")
+
+    # Resource requirements
+    resource_options = parser.add_argument_group(
+        title="Toil options for cores/memory requirements.",
+        description="The options to specify default cores/memory requirements (if not specified by the jobs "
+                    "themselves), and to limit the total amount of memory/cores requested from the batch system."
+    )
+    resource_help_msg = ('The {} amount of {} to request for a job.  '
+                         'Only applicable to jobs that do not specify an explicit value for this requirement.  '
+                         '{}.  '
+                         'Default is {}.')
+    cpu_note = 'Fractions of a core (for example 0.1) are supported on some batch systems [mesos, single_machine]'
+    disk_mem_note = 'Standard suffixes like K, Ki, M, Mi, G or Gi are supported'
+    accelerators_note = (
+        'Each accelerator specification can have a type (gpu [default], nvidia, amd, cuda, rocm, opencl, '
+        'or a specific model like nvidia-tesla-k80), and a count [default: 1]. If both a type and a count '
+        'are used, they must be separated by a colon. If multiple types of accelerators are '
+        'used, the specifications are separated by commas')
+
+    h2b = lambda x: human2bytes(str(x))
+
+    resource_options.add_argument('--defaultMemory', dest='defaultMemory', default="2.0 Gi", type=h2b,
+                                  action=make_open_interval_action(1),
+                                  help=resource_help_msg.format('default', 'memory', disk_mem_note,
+                                                                bytes2human(2147483648)))
+    resource_options.add_argument('--defaultCores', dest='defaultCores', default=1, metavar='FLOAT', type=float,
+                                  action=make_open_interval_action(1.0),
+                                  help=resource_help_msg.format('default', 'cpu', cpu_note, str(1)))
+    resource_options.add_argument('--defaultDisk', dest='defaultDisk', default="2.0 Gi", metavar='INT', type=h2b,
+                                  action=make_open_interval_action(1),
+                                  help=resource_help_msg.format('default', 'disk', disk_mem_note,
+                                                                bytes2human(2147483648)))
+    resource_options.add_argument('--defaultAccelerators', dest='defaultAccelerators', default=[],
+                                  metavar='ACCELERATOR[,ACCELERATOR...]', type=parse_accelerator_list, action="extend",
+                                  help=resource_help_msg.format('default', 'accelerators', accelerators_note, []))
+    resource_options.add_argument('--defaultPreemptible', '--defaultPreemptable', dest='defaultPreemptible',
+                                  metavar='BOOL',
+                                  type=convert_bool, nargs='?', const=True, default=False,
+                                  help='Make all jobs able to run on preemptible (spot) nodes by default.')
+    resource_options.add_argument('--maxCores', dest='maxCores', default=SYS_MAX_SIZE, metavar='INT', type=int,
+                                  action=make_open_interval_action(1),
+                                  help=resource_help_msg.format('max', 'cpu', cpu_note, str(SYS_MAX_SIZE)))
+    resource_options.add_argument('--maxMemory', dest='maxMemory', default=SYS_MAX_SIZE, metavar='INT', type=h2b,
+                                  action=make_open_interval_action(1),
+                                  help=resource_help_msg.format('max', 'memory', disk_mem_note,
+                                                                bytes2human(SYS_MAX_SIZE)))
+    resource_options.add_argument('--maxDisk', dest='maxDisk', default=SYS_MAX_SIZE, metavar='INT', type=h2b,
+                                  action=make_open_interval_action(1),
+                                  help=resource_help_msg.format('max', 'disk', disk_mem_note,
+                                                                bytes2human(SYS_MAX_SIZE)))
+
+    # Retrying/rescuing jobs
+    job_options = parser.add_argument_group(
+        title="Toil options for rescuing/killing/restarting jobs.",
+        description="The options for jobs that either run too long/fail or get lost (some batch systems have issues!)."
+    )
+    job_options.add_argument("--retryCount", dest="retryCount", default=1, type=int,
+                             action=make_open_interval_action(0), metavar="INT",
+                             help=f"Number of times to retry a failing job before giving up and "
+                                  f"labeling job failed. default={1}")
+    job_options.add_argument("--enableUnlimitedPreemptibleRetries", "--enableUnlimitedPreemptableRetries",
+                             dest="enableUnlimitedPreemptibleRetries",
+                             type=convert_bool, default=False, metavar="BOOL",
+                             help="If set, preemptible failures (or any failure due to an instance getting "
+                                  "unexpectedly terminated) will not count towards job failures and --retryCount.")
+    job_options.add_argument("--doubleMem", dest="doubleMem", type=convert_bool, default=False, metavar="BOOL",
+                             help="If set, batch jobs which die to reaching memory limit on batch schedulers "
+                                  "will have their memory doubled and they will be retried. The remaining "
+                                  "retry count will be reduced by 1. Currently supported by LSF.")
+    job_options.add_argument("--maxJobDuration", dest="maxJobDuration", default=SYS_MAX_SIZE, type=int,
+                             action=make_open_interval_action(1), metavar="INT",
+                             help=f"Maximum runtime of a job (in seconds) before we kill it (this is a lower bound, "
+                                  f"and the actual time before killing the job may be longer).  "
+                                  f"default=%(default)s")
+    job_options.add_argument("--rescueJobsFrequency", dest="rescueJobsFrequency", default=60, type=int,
+                             action=make_open_interval_action(1), metavar="INT",
+                             help=f"Period of time to wait (in seconds) between checking for missing/overlong jobs, "
+                                  f"that is jobs which get lost by the batch system. Expert parameter.  "
+                                  f"default=%(default)s")
+
+    # Log management options
+    log_options = parser.add_argument_group(
+        title="Toil log management options.",
+        description="Options for how Toil should manage its logs."
+    )
+    log_options.add_argument("--maxLogFileSize", dest="maxLogFileSize", default=64000, type=h2b,
+                             action=make_open_interval_action(1),
+                             help=f"The maximum size of a job log file to keep (in bytes), log files larger than "
+                                  f"this will be truncated to the last X bytes. Setting this option to zero will "
+                                  f"prevent any truncation. Setting this option to a negative value will truncate "
+                                  f"from the beginning.  Default={bytes2human(64000)}")
+    log_options.add_argument("--writeLogs", dest="writeLogs", nargs='?', action='store', default=None,
+                             const=os.getcwd(), metavar="OPT_PATH",
+                             help="Write worker logs received by the leader into their own files at the specified "
+                                  "path. Any non-empty standard output and error from failed batch system jobs will "
+                                  "also be written into files at this path.  The current working directory will be "
+                                  "used if a path is not specified explicitly. Note: By default only the logs of "
+                                  "failed jobs are returned to leader. Set log level to 'debug' or enable "
+                                  "'--writeLogsFromAllJobs' to get logs back from successful jobs, and adjust "
+                                  "'maxLogFileSize' to control the truncation limit for worker logs.")
+    log_options.add_argument("--writeLogsGzip", dest="writeLogsGzip", nargs='?', action='store', default=None,
+                             const=os.getcwd(), metavar="OPT_PATH",
+                             help="Identical to --writeLogs except the logs files are gzipped on the leader.")
+    log_options.add_argument("--writeLogsFromAllJobs", dest="writeLogsFromAllJobs", type=convert_bool,
+                             default=False, metavar="BOOL",
+                             help="Whether to write logs from all jobs (including the successful ones) without "
+                                  "necessarily setting the log level to 'debug'. Ensure that either --writeLogs "
+                                  "or --writeLogsGzip is set if enabling this option.")
+    log_options.add_argument("--writeMessages", dest="write_messages", default=None,
+                             type=lambda x: None if x is None else os.path.abspath(x), metavar="PATH",
+                             help="File to send messages from the leader's message bus to.")
+    log_options.add_argument("--realTimeLogging", dest="realTimeLogging", type=convert_bool, default=False,
+                             help="Enable real-time logging from workers to leader")
+
+    # Misc options
+    misc_options = parser.add_argument_group(
+        title="Toil miscellaneous options.",
+        description="Everything else."
+    )
+    misc_options.add_argument('--disableChaining', dest='disableChaining', type=convert_bool, default=False,
+                              metavar="BOOL",
+                              help="Disables chaining of jobs (chaining uses one job's resource allocation "
+                                   "for its successor job if possible).")
+    misc_options.add_argument("--disableJobStoreChecksumVerification", dest="disableJobStoreChecksumVerification",
+                              default=False, type=convert_bool, metavar="BOOL",
+                              help="Disables checksum verification for files transferred to/from the job store.  "
+                                   "Checksum verification is a safety check to ensure the data is not corrupted "
+                                   "during transfer. Currently only supported for non-streaming AWS files.")
+
+    class SSEKeyAction(Action):
+        def __call__(self, parser: Any, namespace: Any, values: Any, option_string: Any = None) -> None:
+            if values is not None:
+                sse_key = values
+                if sse_key is None:
+                    return
+                with open(sse_key) as f:
+                    assert len(f.readline().rstrip()) == 32, 'SSE key appears to be invalid.'
+            setattr(namespace, self.dest, values)
+
+    misc_options.add_argument("--sseKey", dest="sseKey", default=None, action=SSEKeyAction, metavar="PATH",
+                              help="Path to file containing 32 character key to be used for server-side encryption on "
+                                   "awsJobStore or googleJobStore. SSE will not be used if this flag is not passed.")
+
+    # yaml.safe_load is being deprecated, this is the suggested workaround
+    def yaml_safe_load(stream: Any) -> Any:
+        yaml = YAML(typ='safe', pure=True)
+        d = yaml.load(stream)
+        if isinstance(d, dict):
+            # this means the argument was a dictionary and is valid yaml (for configargparse)
+            return d
+        else:
+            # this means the argument is likely in it's string format (for CLI)
+            return parse_set_env(parse_str_list(stream))
+
+    class ExtendActionDict(Action):
+        """
+        Argparse action class to implement the action="extend" functionality on dictionaries
+        """
+
+        def __call__(self, parser: Any, namespace: Any, values: Any, option_string: Any = None) -> None:
+            items = getattr(namespace, self.dest, None)
+            assert items is not None  # for mypy. This should never be None, esp. if called in setEnv
+            # note: this will overwrite existing entries
+            items.update(values)
+
+    misc_options.add_argument("--setEnv", '-e', metavar='NAME=VALUE or NAME', dest="environment",
+                              default={}, type=yaml_safe_load, action=ExtendActionDict,
+                              help="Set an environment variable early on in the worker. If VALUE is null, it will "
+                                   "be looked up in the current environment. Independently of this option, the worker "
+                                   "will try to emulate the leader's environment before running a job, except for "
+                                   "some variables known to vary across systems.  Using this option, a variable can "
+                                   "be injected into the worker process itself before it is started.")
+    misc_options.add_argument("--servicePollingInterval", dest="servicePollingInterval", default=60.0, type=float,
+                              action=make_open_interval_action(0.0), metavar="FLOAT",
+                              help=f"Interval of time service jobs wait between polling for the existence of the "
+                                   f"keep-alive flag.  Default: {60.0}")
+    misc_options.add_argument('--forceDockerAppliance', dest='forceDockerAppliance', type=convert_bool, default=False,
+                              metavar="BOOL",
+                              help='Disables sanity checking the existence of the docker image specified by '
+                                   'TOIL_APPLIANCE_SELF, which Toil uses to provision mesos for autoscaling.')
+    misc_options.add_argument('--statusWait', dest='statusWait', type=int, default=3600, metavar="INT",
+                              help="Seconds to wait between reports of running jobs.")
+    misc_options.add_argument('--disableProgress', dest='disableProgress', type=convert_bool, default=False,
+                              metavar="BOOL",
+                              help="Disables the progress bar shown when standard error is a terminal.")
+
+    # Debug options
+    debug_options = parser.add_argument_group(
+        title="Toil debug options.",
+        description="Debug options for finding problems or helping with testing."
+    )
+    debug_options.add_argument("--debugWorker", dest="debugWorker", default=False, action="store_true",
+                               help="Experimental no forking mode for local debugging.  Specifically, workers "
+                                    "are not forked and stderr/stdout are not redirected to the log.")
+    debug_options.add_argument("--disableWorkerOutputCapture", dest="disableWorkerOutputCapture", default=False,
+                               action="store_true",
+                               help="Let worker output go to worker's standard out/error instead of per-job logs.")
+    debug_options.add_argument("--badWorker", dest="badWorker", default=0.0, type=float,
+                               action=make_closed_interval_action(0.0, 1.0), metavar="FLOAT",
+                               help=f"For testing purposes randomly kill --badWorker proportion of jobs using "
+                                    f"SIGKILL.  default={0.0}")
+    debug_options.add_argument("--badWorkerFailInterval", dest="badWorkerFailInterval", default=0.01, type=float,
+                               action=make_open_interval_action(0.0), metavar="FLOAT", # might be cyclical?
+                               help=f"When killing the job pick uniformly within the interval from 0.0 to "
+                                    f"--badWorkerFailInterval seconds after the worker starts.  "
+                                    f"default={0.01}")
+
+    # All deprecated options:
+
+    # These are deprecated in favor of a simpler option
+    #   ex: noLinkImports and linkImports can be simplified into a single link_imports argument
+    link_imports.add_argument("--noLinkImports", dest="linkImports", action="store_false",
+                              help=SUPPRESS)
+    link_imports.add_argument("--linkImports", dest="linkImports", action="store_true",
+                              help=SUPPRESS)
+    link_imports.set_defaults(linkImports=None)
+
+    move_exports.add_argument("--moveExports", dest="moveExports", action="store_true",
+                              help=SUPPRESS)
+    move_exports.add_argument("--noMoveExports", dest="moveExports", action="store_false",
+                              help=SUPPRESS)
+    link_imports.set_defaults(moveExports=None)
+
+    # dest is set to enableCaching to not conflict with the current --caching destination
+    caching.add_argument('--disableCaching', dest='enableCaching', action='store_false', help=SUPPRESS)
+    caching.set_defaults(disableCaching=None)
diff --git a/src/toil/options/cwl.py b/src/toil/options/cwl.py
new file mode 100644
index 00000000..1020dc30
--- /dev/null
+++ b/src/toil/options/cwl.py
@@ -0,0 +1,332 @@
+import os
+from argparse import ArgumentParser
+
+from configargparse import SUPPRESS
+
+from toil.version import baseVersion
+
+
+def add_cwl_options(parser: ArgumentParser, suppress: bool = True) -> None:
+    """
+    Add CWL options to the parser. This only adds nonpositional CWL arguments.
+
+    :param parser: Parser to add options to
+    :param suppress: Suppress help output
+    :return: None
+    """
+    suppress_help = SUPPRESS if suppress else None
+    parser.add_argument("--not-strict", action="store_true", help=suppress_help)
+    parser.add_argument(
+        "--enable-dev",
+        action="store_true",
+        help=suppress_help or suppress_help or "Enable loading and running development versions of CWL",
+    )
+    parser.add_argument(
+        "--enable-ext",
+        action="store_true",
+        help=suppress_help or "Enable loading and running 'cwltool:' extensions to the CWL standards.",
+        default=False,
+    )
+    parser.add_argument("--quiet", dest="quiet", action="store_true", default=False, help=suppress_help)
+    parser.add_argument("--basedir", type=str, help=suppress_help)  # TODO: Might be hard-coded?
+    parser.add_argument("--outdir", type=str, default=None, help=suppress_help)
+    parser.add_argument("--version", action="version", version=baseVersion,
+                        help=suppress_help or "show program's version number and exit")
+    parser.add_argument(
+        "--log-dir",
+        type=str,
+        default="",
+        help=suppress_help or "Log your tools stdout/stderr to this location outside of container",
+    )
+    dockergroup = parser.add_mutually_exclusive_group()
+    dockergroup.add_argument(
+        "--user-space-docker-cmd",
+        help=suppress_help or "(Linux/OS X only) Specify a user space docker command (like "
+                              "udocker or dx-docker) that will be used to call 'pull' and 'run'",
+    )
+    dockergroup.add_argument(
+        "--singularity",
+        action="store_true",
+        default=False,
+        help=suppress_help or "Use Singularity runtime for running containers. "
+                              "Requires Singularity v2.6.1+ and Linux with kernel version v3.18+ or "
+                              "with overlayfs support backported.",
+    )
+    dockergroup.add_argument(
+        "--podman",
+        action="store_true",
+        default=False,
+        help=suppress_help or "Use Podman runtime for running containers. ",
+    )
+    dockergroup.add_argument(
+        "--no-container",
+        action="store_true",
+        help=suppress_help or "Do not execute jobs in a "
+                              "Docker container, even when `DockerRequirement` "
+                              "is specified under `hints`.",
+    )
+    dockergroup.add_argument(
+        "--leave-container",
+        action="store_false",
+        default=True,
+        help=suppress_help or "Do not delete Docker container used by jobs after they exit",
+        dest="rm_container",
+    )
+    extra_dockergroup = parser.add_argument_group()
+    extra_dockergroup.add_argument(
+        "--custom-net",
+        help=suppress_help or "Specify docker network name to pass to docker run command",
+    )
+    cidgroup = parser.add_argument_group(
+        "Options for recording the Docker container identifier into a file."
+    )
+    cidgroup.add_argument(
+        # Disabled as containerid is now saved by default
+        "--record-container-id",
+        action="store_true",
+        default=False,
+        help=suppress_help or SUPPRESS,
+        dest="record_container_id",
+    )
+
+    cidgroup.add_argument(
+        "--cidfile-dir",
+        type=str,
+        help=suppress_help or "Store the Docker container ID into a file in the specified directory.",
+        default=None,
+        dest="cidfile_dir",
+    )
+
+    cidgroup.add_argument(
+        "--cidfile-prefix",
+        type=str,
+        help=suppress_help or "Specify a prefix to the container ID filename. "
+                              "Final file name will be followed by a timestamp. "
+                              "The default is no prefix.",
+        default=None,
+        dest="cidfile_prefix",
+    )
+
+    parser.add_argument(
+        "--preserve-environment",
+        type=str,
+        nargs="+",
+        help=suppress_help or "Preserve specified environment variables when running"
+                              " CommandLineTools",
+        metavar=("VAR1 VAR2"),
+        default=("PATH",),
+        dest="preserve_environment",
+    )
+    parser.add_argument(
+        "--preserve-entire-environment",
+        action="store_true",
+        help=suppress_help or "Preserve all environment variable when running CommandLineTools.",
+        default=False,
+        dest="preserve_entire_environment",
+    )
+    parser.add_argument(
+        "--destBucket",
+        type=str,
+        help=suppress_help or "Specify a cloud bucket endpoint for output files.",
+    )
+    parser.add_argument("--beta-dependency-resolvers-configuration", default=None, help=suppress_help)
+    parser.add_argument("--beta-dependencies-directory", default=None, help=suppress_help)
+    parser.add_argument("--beta-use-biocontainers", default=None, action="store_true", help=suppress_help)
+    parser.add_argument("--beta-conda-dependencies", default=None, action="store_true", help=suppress_help)
+    parser.add_argument(
+        "--tmpdir-prefix",
+        type=str,
+        help=suppress_help or "Path prefix for temporary directories",
+        default=None,
+    )
+    parser.add_argument(
+        "--tmp-outdir-prefix",
+        type=str,
+        help=suppress_help or "Path prefix for intermediate output directories",
+        default=None,
+    )
+    parser.add_argument(
+        "--force-docker-pull",
+        action="store_true",
+        default=False,
+        dest="force_docker_pull",
+        help=suppress_help or "Pull latest docker image even if it is locally present",
+    )
+    parser.add_argument(
+        "--no-match-user",
+        action="store_true",
+        default=False,
+        help=suppress_help or "Disable passing the current uid to `docker run --user`",
+    )
+    parser.add_argument(
+        "--no-read-only",
+        action="store_true",
+        default=False,
+        help=suppress_help or "Do not set root directory in the container as read-only",
+    )
+    parser.add_argument(
+        "--strict-memory-limit",
+        action="store_true",
+        help=suppress_help or "When running with "
+                              "software containers and the Docker engine, pass either the "
+                              "calculated memory allocation from ResourceRequirements or the "
+                              "default of 1 gigabyte to Docker's --memory option.",
+    )
+    parser.add_argument(
+        "--strict-cpu-limit",
+        action="store_true",
+        help=suppress_help or "When running with "
+                              "software containers and the Docker engine, pass either the "
+                              "calculated cpu allocation from ResourceRequirements or the "
+                              "default of 1 core to Docker's --cpu option. "
+                              "Requires docker version >= v1.13.",
+    )
+    parser.add_argument(
+        "--relax-path-checks",
+        action="store_true",
+        default=False,
+        help=suppress_help or "Relax requirements on path names to permit "
+                              "spaces and hash characters.",
+        dest="relax_path_checks",
+    )
+    parser.add_argument(
+        "--default-container",
+        help=suppress_help or "Specify a default docker container that will be "
+                              "used if the workflow fails to specify one.",
+    )
+    parser.add_argument(
+        "--disable-validate",
+        dest="do_validate",
+        action="store_false",
+        default=True,
+        help=suppress_help or SUPPRESS,
+    )
+    parser.add_argument(
+        "--fast-parser",
+        dest="fast_parser",
+        action="store_true",
+        default=False,
+        help=suppress_help or SUPPRESS,
+    )
+    checkgroup = parser.add_mutually_exclusive_group()
+    checkgroup.add_argument(
+        "--compute-checksum",
+        action="store_true",
+        default=True,
+        help=suppress_help or "Compute checksum of contents while collecting outputs",
+        dest="compute_checksum",
+    )
+    checkgroup.add_argument(
+        "--no-compute-checksum",
+        action="store_false",
+        help=suppress_help or "Do not compute checksum of contents while collecting outputs",
+        dest="compute_checksum",
+    )
+
+    parser.add_argument(
+        "--eval-timeout",
+        help=suppress_help or "Time to wait for a Javascript expression to evaluate before giving "
+                              "an error, default 20s.",
+        type=float,
+        default=20,
+    )
+    parser.add_argument(
+        "--overrides",
+        type=str,
+        default=None,
+        help=suppress_help or "Read process requirement overrides from file.",
+    )
+
+    parser.add_argument(
+        "--mpi-config-file",
+        type=str,
+        default=None,
+        help=suppress_help or "Platform specific configuration for MPI (parallel "
+                              "launcher, its flag etc). See the cwltool README "
+                              "section 'Running MPI-based tools' for details of the format: "
+                              "https://github.com/common-workflow-language/cwltool#running-mpi-based-tools-that-need-to-be-launched",
+    )
+    parser.add_argument(
+        "--bypass-file-store",
+        action="store_true",
+        default=False,
+        help=suppress_help or "Do not use Toil's file store and assume all "
+                              "paths are accessible in place from all nodes.",
+        dest="bypass_file_store",
+    )
+    parser.add_argument(
+        "--reference-inputs",
+        action="store_true",
+        default=False,
+        help=suppress_help or "Do not copy remote inputs into Toil's file "
+        "store and assume they are accessible in place from "
+        "all nodes.",
+        dest="reference_inputs",
+    )
+    parser.add_argument(
+        "--disable-streaming",
+        action="store_true",
+        default=False,
+        help=suppress_help or "Disable file streaming for files that have 'streamable' flag True",
+        dest="disable_streaming",
+    )
+
+    provgroup = parser.add_argument_group(
+        "Options for recording provenance information of the execution"
+    )
+    provgroup.add_argument(
+        "--provenance",
+        help=suppress_help or "Save provenance to specified folder as a "
+                              "Research Object that captures and aggregates "
+                              "workflow execution and data products.",
+        type=str,
+    )
+
+    provgroup.add_argument(
+        "--enable-user-provenance",
+        default=False,
+        action="store_true",
+        help=suppress_help or "Record user account info as part of provenance.",
+        dest="user_provenance",
+    )
+    provgroup.add_argument(
+        "--disable-user-provenance",
+        default=False,
+        action="store_false",
+        help=suppress_help or "Do not record user account info in provenance.",
+        dest="user_provenance",
+    )
+    provgroup.add_argument(
+        "--enable-host-provenance",
+        default=False,
+        action="store_true",
+        help=suppress_help or "Record host info as part of provenance.",
+        dest="host_provenance",
+    )
+    provgroup.add_argument(
+        "--disable-host-provenance",
+        default=False,
+        action="store_false",
+        help=suppress_help or "Do not record host info in provenance.",
+        dest="host_provenance",
+    )
+    provgroup.add_argument(
+        "--orcid",
+        help=suppress_help or "Record user ORCID identifier as part of "
+                              "provenance, e.g. https://orcid.org/0000-0002-1825-0097 "
+                              "or 0000-0002-1825-0097. Alternatively the environment variable "
+                              "ORCID may be set.",
+        dest="orcid",
+        default=os.environ.get("ORCID", ""),
+        type=str,
+    )
+    provgroup.add_argument(
+        "--full-name",
+        help=suppress_help or "Record full name of user as part of provenance, "
+                              "e.g. Josiah Carberry. You may need to use shell quotes to preserve "
+                              "spaces. Alternatively the environment variable CWL_FULL_NAME may "
+                              "be set.",
+        dest="cwl_full_name",
+        default=os.environ.get("CWL_FULL_NAME", ""),
+        type=str,
+    )
diff --git a/src/toil/options/wdl.py b/src/toil/options/wdl.py
new file mode 100644
index 00000000..fc094b0b
--- /dev/null
+++ b/src/toil/options/wdl.py
@@ -0,0 +1,32 @@
+from argparse import ArgumentParser
+
+from configargparse import SUPPRESS
+
+
+def add_wdl_options(parser: ArgumentParser, suppress: bool = True) -> None:
+    """
+    Add WDL options to a parser. This only adds nonpositional WDL arguments
+    :param parser: Parser to add options to
+    :param suppress: Suppress help output
+    :return: None
+    """
+    suppress_help = SUPPRESS if suppress else None
+    # include arg names without a wdl specifier if suppress is False
+    # this is to avoid possible duplicate options in custom toil scripts, ex outputFile can be a common argument name
+    output_dialect_arguments = ["--wdlOutputDialect"] + (["--outputDialect"] if not suppress else [])
+    parser.add_argument(*output_dialect_arguments, dest="output_dialect", type=str, default='cromwell',
+                        choices=['cromwell', 'miniwdl'],
+                        help=suppress_help or ("JSON output format dialect. 'cromwell' just returns the workflow's "
+                                               "output values as JSON, while 'miniwdl' nests that under an 'outputs' "
+                                               "key, and includes a 'dir' key where files are written."))
+    output_directory_arguments = ["--wdlOutputDirectory"] + (["--outputDirectory", "-o"] if not suppress else [])
+    parser.add_argument(*output_directory_arguments, dest="output_directory", type=str, default=None,
+                        help=suppress_help or (
+                            "Directory or URI prefix to save output files at. By default a new directory is created "
+                            "in the current directory."))
+    output_file_arguments = ["--wdlOutputFile"] + (["--outputFile", "-m"] if not suppress else [])
+    parser.add_argument(*output_file_arguments, dest="output_file", type=str, default=None,
+                        help=suppress_help or "File or URI to save output JSON to.")
+    reference_inputs_arguments = ["--wdlReferenceInputs"] + (["--referenceInputs"] if not suppress else [])
+    parser.add_argument(*reference_inputs_arguments, dest="reference_inputs", type=bool, default=False,
+                        help=suppress_help or "Pass input files by URL")
diff --git a/src/toil/provisioners/clusterScaler.py b/src/toil/provisioners/clusterScaler.py
index 06140d1c..c31ed5b8 100644
--- a/src/toil/provisioners/clusterScaler.py
+++ b/src/toil/provisioners/clusterScaler.py
@@ -32,7 +32,8 @@ from toil.batchSystems.abstractBatchSystem import (AbstractBatchSystem,
                                                    AbstractScalableBatchSystem,
                                                    NodeInfo)
 from toil.bus import ClusterDesiredSizeMessage, ClusterSizeMessage
-from toil.common import Config, defaultTargetTime
+from toil.common import Config
+from toil.options.common import defaultTargetTime
 from toil.job import JobDescription, ServiceJobDescription
 from toil.lib.conversions import bytes2human, human2bytes
 from toil.lib.retry import old_retry
diff --git a/src/toil/utils/toilConfig.py b/src/toil/utils/toilConfig.py
index b7d77325..1e78c1d2 100644
--- a/src/toil/utils/toilConfig.py
+++ b/src/toil/utils/toilConfig.py
@@ -26,7 +26,8 @@ logger = logging.getLogger(__name__)
 def main() -> None:
     parser = ArgParser()
 
-    parser.add_argument("output", default="config.yaml")
+    parser.add_argument("output", default="config.yaml", help="Filepath to write the config file too. Default=%("
+                                                              "default)s")
     add_logging_options(parser)
     options = parser.parse_args()
     set_logging_from_options(options)
diff --git a/src/toil/wdl/wdltoil.py b/src/toil/wdl/wdltoil.py
index df3943ab..3166ec1b 100755
--- a/src/toil/wdl/wdltoil.py
+++ b/src/toil/wdl/wdltoil.py
@@ -47,14 +47,14 @@ from urllib.parse import quote, unquote, urljoin, urlsplit
 
 import WDL.Error
 import WDL.runtime.config
-from configargparse import ArgParser
+from configargparse import ArgParser, SUPPRESS
 from WDL._util import byte_size_units
 from WDL.CLI import print_error
 from WDL.runtime.backend.docker_swarm import SwarmContainer
 from WDL.runtime.backend.singularity import SingularityContainer
 from WDL.runtime.task_container import TaskContainer
 
-from toil.common import Toil, addOptions
+from toil.common import Toil, addOptions, check_and_create_default_config_file
 from toil.fileStores import FileID
 from toil.fileStores.abstractFileStore import AbstractFileStore
 from toil.job import (AcceleratorRequirement,
@@ -2622,28 +2622,12 @@ def main() -> None:
     """
     A Toil workflow to interpret WDL input files.
     """
+    args = sys.argv[1:]
 
     parser = ArgParser(description='Runs WDL files with toil.')
-    addOptions(parser, jobstore_as_flag=True)
-
-    parser.add_argument("wdl_uri", type=str,
-                        help="WDL document URI")
-    parser.add_argument("inputs_uri", type=str, nargs='?',
-                        help="WDL input JSON URI")
-    parser.add_argument("--input", "-i", dest="inputs_uri", type=str,
-                        help="WDL input JSON URI")
-    parser.add_argument("--outputDialect", dest="output_dialect", type=str, default='cromwell', choices=['cromwell', 'miniwdl'],
-                        help=("JSON output format dialect. 'cromwell' just returns the workflow's output"
-                              "values as JSON, while 'miniwdl' nests that under an 'outputs' key, and "
-                              "includes a 'dir' key where files are written."))
-    parser.add_argument("--outputDirectory", "-o", dest="output_directory", type=str, default=None,
-                        help=("Directory or URI prefix to save output files at. By default a new directory is created in the current directory."))
-    parser.add_argument("--outputFile", "-m", dest="output_file", type=str, default=None,
-                        help="File or URI to save output JSON to.")
-    parser.add_argument("--referenceInputs", dest="reference_inputs", type="bool", default=False, #  type: ignore
-                        help="Pass input files by URL")
-
-    options = parser.parse_args(sys.argv[1:])
+    addOptions(parser, jobstore_as_flag=True, wdl=True)
+
+    options = parser.parse_args(args)
 
     # Make sure we have a jobStore
     if options.jobStore is None:


Test patch:
diff --git a/src/toil/test/provisioners/clusterScalerTest.py b/src/toil/test/provisioners/clusterScalerTest.py
index dc4544ac..7556a0aa 100644
--- a/src/toil/test/provisioners/clusterScalerTest.py
+++ b/src/toil/test/provisioners/clusterScalerTest.py
@@ -27,7 +27,8 @@ from unittest.mock import MagicMock
 from toil.batchSystems.abstractBatchSystem import (AbstractBatchSystem,
                                                    AbstractScalableBatchSystem,
                                                    NodeInfo)
-from toil.common import Config, defaultTargetTime
+from toil.common import Config
+from toil.options.common import defaultTargetTime
 from toil.job import JobDescription
 from toil.lib.conversions import human2bytes as h2b
 from toil.provisioners.abstractProvisioner import AbstractProvisioner, Shape