From 7b67f86426023afff04d3f197645ff61b2307a49 Mon Sep 17 00:00:00 2001 From: Adam Novak Date: Thu, 9 Jul 2026 17:04:58 -0400 Subject: [PATCH 1/5] Add a CWL file we could use as a test for getting back standard error log files for failing commands --- src/toil/test/cwl/fail_with_log.cwl | 70 +++++++++++++++++++++++++++++ 1 file changed, 70 insertions(+) create mode 100644 src/toil/test/cwl/fail_with_log.cwl diff --git a/src/toil/test/cwl/fail_with_log.cwl b/src/toil/test/cwl/fail_with_log.cwl new file mode 100644 index 0000000000..f5277b80c4 --- /dev/null +++ b/src/toil/test/cwl/fail_with_log.cwl @@ -0,0 +1,70 @@ +cwlVersion: v1.2 +class: Workflow + +inputs: + - id: text + type: string + default: "This is a test" + +steps: + hello: + run: + class: CommandLineTool + inputs: + s: string + baseCommand: [bash] + arguments: [ "-c", "echo >&2 '$(inputs.s)' ; exit 1" ] + stdout: output_log.txt + stderr: error_log.txt + outputs: + out_file: + type: File + outputBinding: + glob: output_log.txt + err_file: + type: File + outputBinding: + glob: error_log.txt + in: + - id: s + source: text + out: + - id: out_file + - id: err_file + + count: + run: + class: CommandLineTool + inputs: + in_file: File + baseCommand: [wc] + arguments: [ "-l", $(inputs.in_file) ] + stdout: count.txt + stderr: error_log.txt + outputs: + out_file: + type: File + outputBinding: + glob: count.txt + err_file: + type: File + outputBinding: + glob: error_log.txt + in: + - id: in_file + source: hello/out_file + out: + - id: out_file + - id: err_file + +outputs: + - id: hello_log + type: File + outputSource: hello/err_file + - id: count_log + type: File + outputSource: count/err_file + - id: count_result + type: File + outputSource: count/out_file + From 4784261f3efd45c952be7878c5b6752616799950 Mon Sep 17 00:00:00 2001 From: Adam Novak Date: Thu, 16 Jul 2026 11:30:05 -0400 Subject: [PATCH 2/5] Reorder and document cwltoil.py --- src/toil/cwl/cwltoil.py | 1459 ++++++++++++++------------- src/toil/test/cwl/fail_with_log.cwl | 2 +- 2 files changed, 785 insertions(+), 676 deletions(-) diff --git a/src/toil/cwl/cwltoil.py b/src/toil/cwl/cwltoil.py index 0781af5837..299be73c2f 100644 --- a/src/toil/cwl/cwltoil.py +++ b/src/toil/cwl/cwltoil.py @@ -187,74 +187,17 @@ def cwltoil_was_removed() -> None: ) -# The job object passed into CWLJob and CWLWorkflow -# is a dict mapping to tuple of (key, dict) -# the final dict is derived by evaluating each -# tuple looking up the key in the supplied dict. -# -# This is necessary because Toil jobs return a single value (a dict) -# but CWL permits steps to have multiple output parameters that may -# feed into multiple other steps. This transformation maps the key in the -# output object to the correct key of the input object. - - -class UnresolvedDict(dict[Any, Any]): - """Tag to indicate a dict contains promises that must be resolved.""" - - -# CWL Jobs have inputs and outputs and sometimes we have to do a process of resolution to figure out -# what inputs get values from what outputs. Sometimes, collectively we call these inputs and outputs 'ports'. - - -class SkipNull: - """ - Internal sentinel object. - - Indicates a null value produced by each port of a skipped conditional step. - The CWL 1.2 specification calls for treating this the exactly the same as a - null value. - """ - - -def filter_skip_null(name: str, value: Any) -> Any: - """ - Recursively filter out SkipNull objects from 'value'. - - :param name: Name of port producing this value. - Only used when we find an unhandled null from a conditional step - and we print out a warning. The name allows the user to better - localize which step/port was responsible for the unhandled null. - :param value: port output value object - """ - err_flag = [False] - value = _filter_skip_null(value, err_flag) - if err_flag[0]: - logger.warning( - f"In {name}, SkipNull result found and cast to None. \n" - "You had a conditional step that did not run, " - "but you did not use pickValue to handle the skipped input." - ) - return value - - -def _filter_skip_null(value: Any, err_flag: list[bool]) -> Any: +def simplify_list(maybe_list: Any) -> Any: """ - Private implementation for recursively filtering out SkipNull objects from 'value'. + Turn a length one list loaded by cwltool into a scalar. - :param value: port output value object - :param err_flag: A pass by reference boolean (passed by enclosing in a list) that - allows us to flag, at any level of recursion, that we have - encountered a SkipNull. + Anything else is passed as-is, by reference. """ - match value: - case SkipNull(): - err_flag[0] = True - value = None - case list(val_list): - return [_filter_skip_null(v, err_flag) for v in val_list] - case dict(val_dict): - return {k: _filter_skip_null(v, err_flag) for k, v in val_dict.items()} - return value + if isinstance(maybe_list, MutableSequence): + is_list = aslist(maybe_list) + if len(is_list) == 1: + return is_list[0] + return maybe_list def ensure_no_collisions( @@ -287,6 +230,14 @@ def ensure_no_collisions( seen_names.add(wanted_name) +def get_container_engine(runtime_context: cwltool.context.RuntimeContext) -> str: + if runtime_context.podman: + return "podman" + elif runtime_context.singularity: + return "singularity" + return "docker" + + def try_prepull( cwl_tool_uri: str, runtime_context: cwltool.context.RuntimeContext, batchsystem: str ) -> None: @@ -318,6 +269,94 @@ def try_prepull( logger.info("Prepulling the workflow's containers with Docker...") call_command(["cwl-docker-extract", cwl_tool_uri]) +##### +# +# This section deals with the CWL promise system. +# +# This is not to be confused with the Toil promise system, except under +# fconfusing circumstances. +# +# The CWL language has a lot of machinery around specifying where input values +# come from: there's valueFrom, something called linkMerge, a system for +# specifying default values, all as YAML structures, and then on top of *that* +# there's a way to evaluate string expressions, if the appropriate requirement +# is required. +# +# The implementation side of this in cwltool is all about resolve()-ing things. +# +# We need to reimplement a lot of this because in Toil we only have access to +# information from previous jobs, so we can't use whatever global state +# cwltool's implementations use. +# +##### + +# CWL Jobs have inputs and outputs and sometimes we have to do a process of +# resolution to figure out what inputs get values from what outputs. Sometimes, +# collectively we call these inputs and outputs 'ports'. + +# The job object passed into CWLJob and CWLWorkflow is a dict mapping to tuple +# of (key, dict). The final dict is derived by evaluating each tuple looking up +# the key in the supplied dict. +# +# This is necessary because Toil jobs return a single value (here we use a +# dict) but CWL permits steps to have multiple output parameters that may feed +# into multiple other steps. This transformation maps the key in the output +# object to the correct key of the input object. + +class UnresolvedDict(dict[Any, Any]): + """Tag to indicate a dict contains promises that must be resolved.""" + + +class SkipNull: + """ + Internal sentinel object. + + Indicates a null value produced by each port of a skipped conditional step. + The CWL 1.2 specification calls for treating this the exactly the same as a + null value. + """ + + +def filter_skip_null(name: str, value: Any) -> Any: + """ + Recursively filter out SkipNull objects from 'value'. + + :param name: Name of port producing this value. + Only used when we find an unhandled null from a conditional step + and we print out a warning. The name allows the user to better + localize which step/port was responsible for the unhandled null. + :param value: port output value object + """ + err_flag = [False] + value = _filter_skip_null(value, err_flag) + if err_flag[0]: + logger.warning( + f"In {name}, SkipNull result found and cast to None. \n" + "You had a conditional step that did not run, " + "but you did not use pickValue to handle the skipped input." + ) + return value + + +def _filter_skip_null(value: Any, err_flag: list[bool]) -> Any: + """ + Private implementation for recursively filtering out SkipNull objects from 'value'. + + :param value: port output value object + :param err_flag: A pass by reference boolean (passed by enclosing in a list) that + allows us to flag, at any level of recursion, that we have + encountered a SkipNull. + """ + match value: + case SkipNull(): + err_flag[0] = True + value = None + case list(val_list): + return [_filter_skip_null(v, err_flag) for v in val_list] + case dict(val_dict): + return {k: _filter_skip_null(v, err_flag) for k, v in val_dict.items()} + return value + class Conditional: """ @@ -722,482 +761,283 @@ def resolve_dict_w_promises( return result -def simplify_list(maybe_list: Any) -> Any: - """ - Turn a length one list loaded by cwltool into a scalar. - - Anything else is passed as-is, by reference. - """ - if isinstance(maybe_list, MutableSequence): - is_list = aslist(maybe_list) - if len(is_list) == 1: - return is_list[0] - return maybe_list +##### +# +# In this next section, we override a bunch of cwltool classes and put Toil +# hooks all over them. +# +# cwltool has a lot of moving parts here. See +# +# for some information on them. +# +# There's the "tool", a descendant of cwltool.process.Process, which could be a +# cwltool.command_line_tool.CommandLineTool or a +# cwltool.command_line_tool.ExpressionTool. This corresponds to the .cwl file. +# +# There's a "job order", sometimes (in Toil's code to orchestrate the workflow) +# just called the "cwl job", which corresponds to the YAML/JSON inputs that we +# want to run on. +# +# The tool Process also has a .job() method, which yields any of a few types +# (see +# ) +# like cwltool.command_line_tool.CommandLineJob or +# cwltool.command_line_tool.ExpressionJob. Each of these types has a .run() +# method. Despite the confusion this causes with e.g. CWLJob and the "CWL job +# order", these objects are also called "CWL jobs". +# +# The Process gets used in combination with an "executor" descended from +# cwltool.executors.JobExecutor. Toil uses a +# cwltool.executors.SingleJobExecutor descendant. This has a .run_jobs() that +# hands the job order to the Process's .job() to get a CWL job, and then, after +# some more ceremony, runs it via its .run() method. In theory there can be +# multiple CWL jobs kicking around, but Toil only uses Processes that produce +# and executors that expect just one job. +# +# There are a lot of cwltool extension points here, but they're all about +# letting you replace the classes uses by overriding factory functions. We +# really want to get inside the giant methods cwltool defines that do the work +# of setting up, running, and cleaning up afterwards, all in one method, and +# cwltool doesn't have good extension points for that. So mostly we provide our +# own classes for things and then those classes try to patch their way into +# random methods that happen to run at the right times in cwltool's flow in +# order to affect its behavior. +# +##### +##### +# Our root hook is to hook the function that picks what Process type to use to represent a .cwl file. +##### -class ToilPathMapper(PathMapper): +def toil_make_tool( + toolpath_object: CommentedMap, + loadingContext: cwltool.context.LoadingContext, +) -> Process: """ - Keeps track of files in a Toil way. + Factory function that uses Toil-hooked tool classes to load CWL tools. - Maps between the symbolic identifier of a file (the Toil FileID), its local - path on the host (the value returned by readGlobalFile) and the - location of the file inside the software container. + This factory function is meant to be passed to cwltool.load_tool(). """ + if isinstance(toolpath_object, Mapping): + if toolpath_object.get("class") == "CommandLineTool": + return ToilCommandLineTool(toolpath_object, loadingContext) + elif toolpath_object.get("class") == "ExpressionTool": + return ToilExpressionTool(toolpath_object, loadingContext) + return cwltool.workflow.default_make_tool(toolpath_object, loadingContext) - def __init__( - self, - referenced_files: MutableSequence[CWLFileType | CWLDirectoryType], - basedir: str, - stagedir: str, - separateDirs: bool = True, - get_file: Any | None = None, - stage_listing: bool = False, - streaming_allowed: bool = True, - ): +##### +# Then we define a base Toil-flavored CWL Process, and command-line and expression variants of it. +# This goes on to install our PathMapper hooks. +##### + +class ToilTool(Process): + """Mixin to hook Toil into a cwltool tool type.""" + + def __init__(self, *args: Any, **kwargs: Any) -> None: """ - Initialize this ToilPathMapper. + Init hook to set up member variables. + """ + super().__init__(*args, **kwargs) + # Reserve a spot for the Toil job that ends up executing this tool. + self._toil_job: Job | None = None + # Remember path mappers we have used so we can interrogate them later to find out what the job mapped. + self._path_mappers: list[cwltool.pathmapper.PathMapper] = [] - :param stage_listing: Stage files and directories inside directories - even if we also stage the parent. - :param get_file: A function that takes a URL, an optional "streamable" - flag for if a file is supposed to be streamable, and an optional - "streaming_allowed" flag for whether we are running with - streaming on, and returns a file: URI to where the file or - directory has been downloaded to. Meant to be a partially-bound - version of toil_get_file(). - :param referenced_files: List of CWL File and Directory objects, which can have their locations set as both - virtualized and absolute local paths + def connect_toil_job(self, job: Job) -> None: """ - self.get_file = get_file - self.stage_listing = stage_listing - self.streaming_allowed = streaming_allowed + Attach the Toil tool to the Toil job that is executing it. This allows + it to use the Toil job to stop at certain points if debugging flags are + set. + """ + self._toil_job = job - super().__init__(referenced_files, basedir, stagedir, separateDirs=separateDirs) - def visit( + def make_path_mapper( self, - obj: CWLFileType | CWLDirectoryType, + reffiles: MutableSequence[CWLFileType | CWLDirectoryType], stagedir: str, - basedir: str, - copy: bool = False, - staged: bool = False, - ) -> None: - """ - Iterate over a CWL object, resolving File and Directory path references. + runtimeContext: cwltool.context.RuntimeContext, + separateDirs: bool, + ) -> cwltool.pathmapper.PathMapper: + """Create the appropriate PathMapper for the situation.""" + if getattr(runtimeContext, "bypass_file_store", False): + # We only need to understand cwltool's supported URIs + mapper = PathMapper( + reffiles, runtimeContext.basedir, stagedir, separateDirs=separateDirs + ) + else: + # We need to be able to read from Toil-provided URIs + mapper = ToilPathMapper( + reffiles, + runtimeContext.basedir, + stagedir, + separateDirs, + get_file=getattr(runtimeContext, "toil_get_file", None), + streaming_allowed=runtimeContext.streaming_allowed, + ) - This is called on each File or Directory CWL object. The Files and - Directories all have "location" fields. For the Files, these are from - upload_file(), and for the Directories, these are from - upload_directory() or cwltool internally. With upload_directory(), they and their children will be assigned - locations based on listing the Directories using ToilFsAccess. With cwltool, locations will be set as absolute - paths. + # Remember the path mappers + self._path_mappers.append(mapper) + return mapper - :param obj: The CWL File or Directory to process + def __str__(self) -> str: + """Return string representation of this tool type.""" + return f'{self.__class__.__name__}({repr(getattr(self, "tool", {}).get("id", "???"))})' - :param stagedir: The base path for target paths to be generated under, - except when a File or Directory has an overriding parent directory in - dirname - - :param basedir: The directory from which relative paths should be - resolved; used as the base directory for the StdFsAccess that generated - the listing being processed. - - :param copy: If set, use writable types for Files and Directories. +class ToilCommandLineTool(ToilTool, cwltool.command_line_tool.CommandLineTool): + """ + CWL CommandLineTool that connects to ToilPathMapper and other hooks. + """ - :param staged: Starts as True at the top of the recursion. Set to False - when entering a directory that we can actually download, so we don't - stage files and subdirectories separately from the directory as a - whole. Controls the staged flag on generated mappings, and therefore - whether files and directories are actually placed at their mapped-to - target locations. If stage_listing is True, we will leave this True - throughout and stage everything. + def make_job_runner( + self, runtimeContext: cwltool.context.RuntimeContext + ) -> type[cwltool.job.JobBase]: + """ + Use a job class that monitors container resource usage if needed. - Produces one MapperEnt for every unique location for a File or - Directory. These MapperEnt objects are instructions to cwltool's - stage_files function: - https://github.com/common-workflow-language/cwltool/blob/a3e3a5720f7b0131fa4f9c0b3f73b62a347278a6/cwltool/process.py#L254 + We only act on container systems where the container isn't a descendant + process of us. + """ + parent_class = super().make_job_runner(runtimeContext) + if parent_class is DockerCommandLineJob: + # There's only one daemon-containerized container system: it's + # Docker itself (and not any of the subclasses of this), + return ToilDockerCommandLineJob + return parent_class - The MapperEnt has fields: + def _initialworkdir( + self, j: cwltool.job.JobBase | None, builder: cwltool.builder.Builder + ) -> None: + """ + Hook the InitialWorkDirRequirement setup to make sure that there are no + name conflicts at the top level of the work directory and support + Toil's files downloaded hook. + """ - resolved: An absolute local path anywhere on the filesystem where the - file/directory can be found, or the contents of a file to populate it - with if type is CreateWritableFile or CreateFile. Or, a URI understood - by the StdFsAccess in use (for example, toilfile:). + # Set up the initial work dir with all its files + super()._initialworkdir(j, builder) - target: An absolute path under stagedir that the file or directory will - then be placed at by cwltool. Except if a File or Directory has a - dirname field, giving its parent path, that is used instead. + if j is None: + return # Only testing - type: One of: + # The initial work dir listing is now in j.generatefiles["listing"] + # Also j.generatefiles is a CWL Directory. + # So check the initial working directory. + logger.debug("Initial work dir: %s", j.generatefiles) + ensure_no_collisions( + j.generatefiles, + "the job's working directory as specified by the InitialWorkDirRequirement", + ) - File: cwltool will copy or link the file from resolved to target, - if possible. + if self._toil_job is not None: + # Make a table of all the places we mapped files to when downloading the inputs. - CreateFile: cwltool will create the file at target, treating - resolved as the contents. + # We want to hint which host paths and container (if any) paths correspond + host_and_job_paths: list[tuple[str, str]] = [] - WritableFile: cwltool will copy the file from resolved to target, - making it writable. + for pm in self._path_mappers: + for _, mapper_entry in pm.items_exclude_children(): + # We know that mapper_entry.target as seen by the task is + # mapper_entry.resolved on the host. + host_and_job_paths.append( + (mapper_entry.resolved, mapper_entry.target) + ) - CreateWritableFile: cwltool will create the file at target, - treating resolved as the contents, and make it writable. + # Notice that we have downloaded our inputs. Explain which files + # those are here and what the task will expect to call them. + self._toil_job.files_downloaded_hook(host_and_job_paths) - Directory: cwltool will copy or link the directory from resolved to - target, if possible. Otherwise, cwltool will make the directory at - target if resolved starts with "_:". Otherwise it will do nothing. - WritableDirectory: cwltool will copy the directory from resolved to - target, if possible. Otherwise, cwltool will make the directory at - target if resolved starts with "_:". Otherwise it will do nothing. +class ToilExpressionTool(ToilTool, cwltool.command_line_tool.ExpressionTool): + """Subclass the cwltool expression tool to provide the custom ToilPathMapper.""" + # TODO: Don't expression tools also want the resource monitoring hooks? - staged: if set to False, cwltool will not make or copy anything for this entry - """ - logger.debug( - "ToilPathMapper mapping into %s from %s for: %s", stagedir, basedir, obj - ) +##### +# Our other entry point into cwltool for customizing running things is by +# providing our own executor. We construct this ourselves and then use it to +# run our also-customized Processes. +##### - # If the file has a dirname set, we can try and put it there instead of - # wherever else we would stage it. - # TODO: why would we do that? - stagedir = cast(Optional[str], obj.get("dirname")) or stagedir +class ToilSingleJobExecutor(cwltool.executors.SingleJobExecutor): + """ + A SingleJobExecutor that does not assume it is at the top level of the workflow. - if obj["class"] not in ("File", "Directory"): - # We only handle files and directories; only they have locations. - return + We need this because otherwise every job thinks it is top level and tries + to discover secondary files, which may exist when they haven't actually + been passed at the top level and thus aren't supposed to be visible. + """ - location = obj["location"] - if location in self: - # If we've already mapped this, map it consistently. - tgt = self._pathmap[location].target - logger.debug( - "ToilPathMapper re-using target %s for path %s", - tgt, - location, + def run_jobs( + self, + process: Process, + job_order_object: CWLObjectType, + logger: logging.Logger, + runtime_context: cwltool.context.RuntimeContext, + ) -> None: + """run_jobs from SingleJobExecutor, but not in a top level runtime context.""" + runtime_context.toplevel = False + if isinstance( + process, cwltool.command_line_tool.CommandLineTool + ) and isinstance( + process.make_job_runner(runtime_context), SingularityCommandLineJob + ): + # Set defaults for singularity cache environment variables, similar to what we do in wdltoil + # Use the same place as the default singularity cache directory + singularity_cache = os.path.join(os.path.expanduser("~"), ".singularity") + os.environ["SINGULARITY_CACHEDIR"] = os.environ.get( + "SINGULARITY_CACHEDIR", singularity_cache ) - else: - # Decide where to put the file or directory, as an absolute path. - tgt = os.path.join( - stagedir, - obj["basename"], + + # If singularity is detected, prepull the image to ensure locking + docker_req, docker_is_req = process.get_requirement( + feature="DockerRequirement" ) - if self.reversemap(tgt) is not None: - # If the target already exists in the pathmap, but we haven't yet - # mapped this, it means we have a conflict. - i = 2 - new_tgt = f"{tgt}_{i}" - while self.reversemap(new_tgt) is not None: - i += 1 - new_tgt = f"{tgt}_{i}" - logger.debug( - "ToilPathMapper resolving mapping conflict: %s is now %s", - tgt, - new_tgt, + with global_mutex( + os.environ["SINGULARITY_CACHEDIR"], "toil_singularity_cache_mutex" + ): + SingularityCommandLineJob.get_image( + dockerRequirement=cast(dict[str, str], docker_req), + pull_image=runtime_context.pull_image, + force_pull=runtime_context.force_docker_pull, + tmp_outdir_prefix=runtime_context.tmp_outdir_prefix, ) - tgt = new_tgt - if is_directory(obj): - # Whether or not we've already mapped this path, we need to map all - # children recursively. + return super().run_jobs(process, job_order_object, logger, runtime_context) - logger.debug("ToilPathMapper visiting directory %s", location) +##### +# The customized Process objects install hooks to use therse customized CWL job +# objects, which customize either their .run() method or things we know it goes +# on to call. +##### - # We want to check the directory to make sure it is not - # self-contradictory in its immediate children and their names. - ensure_no_collisions(obj) +# We need to know which cwltool container implementations run the container +# under the calling process's child tree. Unlisted container implementations +# are assumed to run it elsewhere through a deamon. +# +# We have to list it this way around (and not list the implementations that use +# a daemon) because some of these extend DockerCommandLineJob. +CHILD_PROCESS_CONTAINER_JOBS = (PodmanCommandLineJob, SingularityCommandLineJob, UDockerCommandLineJob) - # We may need to copy this directory even if we don't copy things inside it. - copy_here = False - # Try and resolve the location to a local path - if location.startswith("file://"): - # This is still from the local machine, so go find where it is - resolved = schema_salad.ref_resolver.uri_file_path(location) - elif location.startswith("toildir:"): - # We need to download this directory (or subdirectory) - if self.get_file: - # We can actually go get it and its contents - resolved = schema_salad.ref_resolver.uri_file_path( - self.get_file(location) - ) - else: - # We are probably staging final outputs on the leader. We - # can't go get the directory. Just pass it through. - resolved = location - elif location.startswith("_:"): - # cwltool made this up for an empty/synthetic directory it - # wants to make. +class ToilDockerCommandLineJob(DockerCommandLineJob): + """Container job that collects resource stats from injected in-container code.""" - # If we let cwltool make the directory and stage it, and then - # stage files inside it, we can end up with Docker creating - # root-owned files in whatever we mounted for the Docker work - # directory, somehow. So make a directory ourselves instead. - if self.get_file: - # Ask for an empty directory - new_dir_uri = self.get_file("_:") - # And get a path for it - resolved = schema_salad.ref_resolver.uri_file_path(new_dir_uri) + def _execute( + self, + runtime: list[str], + env: MutableMapping[str, str], + runtimeContext: cwltool.context.RuntimeContext, + monitor_function: Callable[["subprocess.Popen[str]"], None] | None = None, + ) -> None: + # This job is going to run in a container that makes it not be a + # descendant of our process. So we need to inject code to count its + # resource usage. - if is_directory(obj) and obj["listing"] != []: - # If there's stuff inside here to stage, we need to copy - # this directory here, because we can't Docker mount things - # over top of immutable directories. - copy_here = True - else: - # We can't really make the directory. Maybe we are - # exporting from the leader and it doesn't matter. - resolved = location - elif location.startswith("/"): - # Test if path is an absolute local path - # Does not check if the path is relative - # While Toil encodes paths into a URL with ToilPathMapper, - # something called internally in cwltool may return an absolute path - # ex: if cwltool calls itself internally in command_line_tool.py, - # it collects outputs with collect_output, and revmap_file will use its own internal pathmapper - resolved = location - else: - raise RuntimeError("Unsupported location: " + location) - - if location in self._pathmap: - # Don't map the same directory twice - logger.debug( - "ToilPathMapper stopping recursion because we have already " - "mapped directory: %s", - location, - ) - return - - logger.debug( - "ToilPathMapper adding directory mapping %s -> %s", resolved, tgt - ) - self._pathmap[location] = MapperEnt( - resolved, - tgt, - "WritableDirectory" if (copy or copy_here) else "Directory", - staged, - ) - - if not location.startswith("_:") and not self.stage_listing: - # Don't stage anything below here separately, since we are able - # to copy the whole directory from somewhere and and we can't - # stage files over themselves. - staged = False - - # Keep recursing - self.visitlisting( - obj.get("listing", []), - tgt, - basedir, - copy=copy, - staged=staged, - ) - - if is_file(obj): - logger.debug("ToilPathMapper visiting file %s", location) - - if location in self._pathmap: - # Don't map the same file twice - logger.debug( - "ToilPathMapper stopping recursion because we have already " - "mapped file: %s", - location, - ) - return - - ab = abspath(location, basedir) - if "contents" in obj and location.startswith("_:"): - # We are supposed to create this file - self._pathmap[location] = MapperEnt( - obj["contents"], - tgt, - "CreateWritableFile" if copy else "CreateFile", - staged, - ) - else: - with SourceLine( - obj, - "location", - ValidationException, - logger.isEnabledFor(logging.DEBUG), - ): - # If we have access to the Toil file store, we will have a - # get_file set, and it will convert this path to a file: - # URI for a local file it downloaded. - if self.get_file: - deref = self.get_file( - location, - obj.get("streamable", False), - self.streaming_allowed, - ) - else: - deref = ab - if deref.startswith("file:"): - deref = schema_salad.ref_resolver.uri_file_path(deref) - if urlsplit(deref).scheme in ["http", "https"]: - deref = downloadHttpFile(location) - elif urlsplit(deref).scheme != "toilfile": - # Dereference symbolic links - st = os.lstat(deref) - while stat.S_ISLNK(st.st_mode): - logger.debug("ToilPathMapper following symlink %s", deref) - rl = os.readlink(deref) - deref = ( - rl - if os.path.isabs(rl) - else os.path.join(os.path.dirname(deref), rl) - ) - st = os.lstat(deref) - - # If we didn't download something that is a toilfile: - # reference, we just pass that along. - - """Link or copy files to their targets. Create them as needed.""" - - logger.debug( - "ToilPathMapper adding file mapping %s -> %s", deref, tgt - ) - - self._pathmap[location] = MapperEnt( - deref, tgt, "WritableFile" if copy else "File", staged - ) - - # Handle all secondary files that need to be next to this one. - self.visitlisting( - obj.get("secondaryFiles", []), - stagedir, - basedir, - copy=copy, - staged=staged, - ) - - -class ToilSingleJobExecutor(cwltool.executors.SingleJobExecutor): - """ - A SingleJobExecutor that does not assume it is at the top level of the workflow. - - We need this because otherwise every job thinks it is top level and tries - to discover secondary files, which may exist when they haven't actually - been passed at the top level and thus aren't supposed to be visible. - """ - - def run_jobs( - self, - process: Process, - job_order_object: CWLObjectType, - logger: logging.Logger, - runtime_context: cwltool.context.RuntimeContext, - ) -> None: - """run_jobs from SingleJobExecutor, but not in a top level runtime context.""" - runtime_context.toplevel = False - if isinstance( - process, cwltool.command_line_tool.CommandLineTool - ) and isinstance( - process.make_job_runner(runtime_context), SingularityCommandLineJob - ): - # Set defaults for singularity cache environment variables, similar to what we do in wdltoil - # Use the same place as the default singularity cache directory - singularity_cache = os.path.join(os.path.expanduser("~"), ".singularity") - os.environ["SINGULARITY_CACHEDIR"] = os.environ.get( - "SINGULARITY_CACHEDIR", singularity_cache - ) - - # If singularity is detected, prepull the image to ensure locking - docker_req, docker_is_req = process.get_requirement( - feature="DockerRequirement" - ) - with global_mutex( - os.environ["SINGULARITY_CACHEDIR"], "toil_singularity_cache_mutex" - ): - SingularityCommandLineJob.get_image( - dockerRequirement=cast(dict[str, str], docker_req), - pull_image=runtime_context.pull_image, - force_pull=runtime_context.force_docker_pull, - tmp_outdir_prefix=runtime_context.tmp_outdir_prefix, - ) - - return super().run_jobs(process, job_order_object, logger, runtime_context) - - -class ToilTool: - """Mixin to hook Toil into a cwltool tool type.""" - - def __init__(self, *args: Any, **kwargs: Any) -> None: - """ - Init hook to set up member variables. - """ - super().__init__(*args, **kwargs) - # Reserve a spot for the Toil job that ends up executing this tool. - self._toil_job: Job | None = None - # Remember path mappers we have used so we can interrogate them later to find out what the job mapped. - self._path_mappers: list[cwltool.pathmapper.PathMapper] = [] - - def connect_toil_job(self, job: Job) -> None: - """ - Attach the Toil tool to the Toil job that is executing it. This allows - it to use the Toil job to stop at certain points if debugging flags are - set. - """ - self._toil_job = job - - def make_path_mapper( - self, - reffiles: MutableSequence[CWLFileType | CWLDirectoryType], - stagedir: str, - runtimeContext: cwltool.context.RuntimeContext, - separateDirs: bool, - ) -> cwltool.pathmapper.PathMapper: - """Create the appropriate PathMapper for the situation.""" - if getattr(runtimeContext, "bypass_file_store", False): - # We only need to understand cwltool's supported URIs - mapper = PathMapper( - reffiles, runtimeContext.basedir, stagedir, separateDirs=separateDirs - ) - else: - # We need to be able to read from Toil-provided URIs - mapper = ToilPathMapper( - reffiles, - runtimeContext.basedir, - stagedir, - separateDirs, - get_file=getattr(runtimeContext, "toil_get_file", None), - streaming_allowed=runtimeContext.streaming_allowed, - ) - - # Remember the path mappers - self._path_mappers.append(mapper) - return mapper - - def __str__(self) -> str: - """Return string representation of this tool type.""" - return f'{self.__class__.__name__}({repr(getattr(self, "tool", {}).get("id", "???"))})' - -# We need to know which cwltool container implementations run the container -# under the calling process's child tree. Unlisted container implementations -# are assumed to run it elsewhere through a deamon. -# -# We have to list it this way around (and not list the implementations that use -# a daemon) because some of these extend DockerCommandLineJob. -CHILD_PROCESS_CONTAINER_JOBS = (PodmanCommandLineJob, SingularityCommandLineJob, UDockerCommandLineJob) - - -class ToilDockerCommandLineJob(DockerCommandLineJob): - """Container job that collects resource stats from injected in-container code.""" - - def _execute( - self, - runtime: list[str], - env: MutableMapping[str, str], - runtimeContext: cwltool.context.RuntimeContext, - monitor_function: Callable[["subprocess.Popen[str]"], None] | None = None, - ) -> None: - # This job is going to run in a container that makes it not be a - # descendant of our process. So we need to inject code to count its - # resource usage. - - # First we need the command that's actually going to run. To be fully - # spec-compliant, we need to make sure Docker ENTRYPOINT keeps working. - # So we need to fetch it out, put it in the script, and remember to - # override it with a shell later. See - # + # First we need the command that's actually going to run. To be fully + # spec-compliant, we need to make sure Docker ENTRYPOINT keeps working. + # So we need to fetch it out, put it in the script, and remember to + # override it with a shell later. See + # # We know the selected container image is the last element in runtime. # Inspect it to get its entrypoint list @@ -1278,89 +1118,349 @@ def _file_mounts_from_pathmapper(self) -> list[tuple[str, str]]: file_mounts.append((ent.resolved, ent.target)) return file_mounts +##### +# +# In this section, we define the classes for hooking CWL's file management into +# Toil's FileStore. This is mostly based around a PathMapper implementation and +# an StdFsAccess impolementation. Since cwltool doesn't provide very specific +# requirements for what these implementations need to guarantee, a lot of this +# has been learned by trial and error +# +##### -class ToilCommandLineTool(ToilTool, cwltool.command_line_tool.CommandLineTool): - """Subclass the cwltool command line tool to provide the custom ToilPathMapper - and add the monitoring code to the job's container command line.""" +class ToilPathMapper(PathMapper): + """ + Keeps track of files in a Toil way. - def make_job_runner( - self, runtimeContext: cwltool.context.RuntimeContext - ) -> type[cwltool.job.JobBase]: - """ - Use a job class that monitors container resource usage if needed. + Maps between the symbolic identifier of a file (the Toil FileID), its local + path on the host (the value returned by readGlobalFile) and the + location of the file inside the software container. + """ - We only act on container systems where the container isn't a descendant - process of us. - """ - parent_class = super().make_job_runner(runtimeContext) - if parent_class is DockerCommandLineJob: - # There's only one daemon-containerized container system: it's - # Docker itself (and not any of the subclasses of this), - return ToilDockerCommandLineJob - return parent_class + def __init__( + self, + referenced_files: MutableSequence[CWLFileType | CWLDirectoryType], + basedir: str, + stagedir: str, + separateDirs: bool = True, + get_file: Any | None = None, + stage_listing: bool = False, + streaming_allowed: bool = True, + ): + """ + Initialize this ToilPathMapper. - def _initialworkdir( - self, j: cwltool.job.JobBase | None, builder: cwltool.builder.Builder - ) -> None: + :param stage_listing: Stage files and directories inside directories + even if we also stage the parent. + :param get_file: A function that takes a URL, an optional "streamable" + flag for if a file is supposed to be streamable, and an optional + "streaming_allowed" flag for whether we are running with + streaming on, and returns a file: URI to where the file or + directory has been downloaded to. Meant to be a partially-bound + version of toil_get_file(). + :param referenced_files: List of CWL File and Directory objects, which can have their locations set as both + virtualized and absolute local paths """ - Hook the InitialWorkDirRequirement setup to make sure that there are no - name conflicts at the top level of the work directory. + self.get_file = get_file + self.stage_listing = stage_listing + self.streaming_allowed = streaming_allowed + + super().__init__(referenced_files, basedir, stagedir, separateDirs=separateDirs) + + def visit( + self, + obj: CWLFileType | CWLDirectoryType, + stagedir: str, + basedir: str, + copy: bool = False, + staged: bool = False, + ) -> None: """ + Iterate over a CWL object, resolving File and Directory path references. - # Set up the initial work dir with all its files - super()._initialworkdir(j, builder) + This is called on each File or Directory CWL object. The Files and + Directories all have "location" fields. For the Files, these are from + upload_file(), and for the Directories, these are from + upload_directory() or cwltool internally. With upload_directory(), they and their children will be assigned + locations based on listing the Directories using ToilFsAccess. With cwltool, locations will be set as absolute + paths. - if j is None: - return # Only testing + :param obj: The CWL File or Directory to process - # The initial work dir listing is now in j.generatefiles["listing"] - # Also j.generatefiles is a CWL Directory. - # So check the initial working directory. - logger.debug("Initial work dir: %s", j.generatefiles) - ensure_no_collisions( - j.generatefiles, - "the job's working directory as specified by the InitialWorkDirRequirement", + :param stagedir: The base path for target paths to be generated under, + except when a File or Directory has an overriding parent directory in + dirname + + :param basedir: The directory from which relative paths should be + resolved; used as the base directory for the StdFsAccess that generated + the listing being processed. + + :param copy: If set, use writable types for Files and Directories. + + :param staged: Starts as True at the top of the recursion. Set to False + when entering a directory that we can actually download, so we don't + stage files and subdirectories separately from the directory as a + whole. Controls the staged flag on generated mappings, and therefore + whether files and directories are actually placed at their mapped-to + target locations. If stage_listing is True, we will leave this True + throughout and stage everything. + + Produces one MapperEnt for every unique location for a File or + Directory. These MapperEnt objects are instructions to cwltool's + stage_files function: + https://github.com/common-workflow-language/cwltool/blob/a3e3a5720f7b0131fa4f9c0b3f73b62a347278a6/cwltool/process.py#L254 + + The MapperEnt has fields: + + resolved: An absolute local path anywhere on the filesystem where the + file/directory can be found, or the contents of a file to populate it + with if type is CreateWritableFile or CreateFile. Or, a URI understood + by the StdFsAccess in use (for example, toilfile:). + + target: An absolute path under stagedir that the file or directory will + then be placed at by cwltool. Except if a File or Directory has a + dirname field, giving its parent path, that is used instead. + + type: One of: + + File: cwltool will copy or link the file from resolved to target, + if possible. + + CreateFile: cwltool will create the file at target, treating + resolved as the contents. + + WritableFile: cwltool will copy the file from resolved to target, + making it writable. + + CreateWritableFile: cwltool will create the file at target, + treating resolved as the contents, and make it writable. + + Directory: cwltool will copy or link the directory from resolved to + target, if possible. Otherwise, cwltool will make the directory at + target if resolved starts with "_:". Otherwise it will do nothing. + + WritableDirectory: cwltool will copy the directory from resolved to + target, if possible. Otherwise, cwltool will make the directory at + target if resolved starts with "_:". Otherwise it will do nothing. + + staged: if set to False, cwltool will not make or copy anything for this entry + """ + logger.debug( + "ToilPathMapper mapping into %s from %s for: %s", stagedir, basedir, obj ) - if self._toil_job is not None: - # Make a table of all the places we mapped files to when downloading the inputs. + # If the file has a dirname set, we can try and put it there instead of + # wherever else we would stage it. + # TODO: why would we do that? + stagedir = cast(Optional[str], obj.get("dirname")) or stagedir - # We want to hint which host paths and container (if any) paths correspond - host_and_job_paths: list[tuple[str, str]] = [] + if obj["class"] not in ("File", "Directory"): + # We only handle files and directories; only they have locations. + return - for pm in self._path_mappers: - for _, mapper_entry in pm.items_exclude_children(): - # We know that mapper_entry.target as seen by the task is - # mapper_entry.resolved on the host. - host_and_job_paths.append( - (mapper_entry.resolved, mapper_entry.target) + location = obj["location"] + if location in self: + # If we've already mapped this, map it consistently. + tgt = self._pathmap[location].target + logger.debug( + "ToilPathMapper re-using target %s for path %s", + tgt, + location, + ) + else: + # Decide where to put the file or directory, as an absolute path. + tgt = os.path.join( + stagedir, + obj["basename"], + ) + if self.reversemap(tgt) is not None: + # If the target already exists in the pathmap, but we haven't yet + # mapped this, it means we have a conflict. + i = 2 + new_tgt = f"{tgt}_{i}" + while self.reversemap(new_tgt) is not None: + i += 1 + new_tgt = f"{tgt}_{i}" + logger.debug( + "ToilPathMapper resolving mapping conflict: %s is now %s", + tgt, + new_tgt, + ) + tgt = new_tgt + + if is_directory(obj): + # Whether or not we've already mapped this path, we need to map all + # children recursively. + + logger.debug("ToilPathMapper visiting directory %s", location) + + # We want to check the directory to make sure it is not + # self-contradictory in its immediate children and their names. + ensure_no_collisions(obj) + + # We may need to copy this directory even if we don't copy things inside it. + copy_here = False + + # Try and resolve the location to a local path + if location.startswith("file://"): + # This is still from the local machine, so go find where it is + resolved = schema_salad.ref_resolver.uri_file_path(location) + elif location.startswith("toildir:"): + # We need to download this directory (or subdirectory) + if self.get_file: + # We can actually go get it and its contents + resolved = schema_salad.ref_resolver.uri_file_path( + self.get_file(location) ) + else: + # We are probably staging final outputs on the leader. We + # can't go get the directory. Just pass it through. + resolved = location + elif location.startswith("_:"): + # cwltool made this up for an empty/synthetic directory it + # wants to make. - # Notice that we have downloaded our inputs. Explain which files - # those are here and what the task will expect to call them. - self._toil_job.files_downloaded_hook(host_and_job_paths) - + # If we let cwltool make the directory and stage it, and then + # stage files inside it, we can end up with Docker creating + # root-owned files in whatever we mounted for the Docker work + # directory, somehow. So make a directory ourselves instead. + if self.get_file: + # Ask for an empty directory + new_dir_uri = self.get_file("_:") + # And get a path for it + resolved = schema_salad.ref_resolver.uri_file_path(new_dir_uri) + + if is_directory(obj) and obj["listing"] != []: + # If there's stuff inside here to stage, we need to copy + # this directory here, because we can't Docker mount things + # over top of immutable directories. + copy_here = True + else: + # We can't really make the directory. Maybe we are + # exporting from the leader and it doesn't matter. + resolved = location + elif location.startswith("/"): + # Test if path is an absolute local path + # Does not check if the path is relative + # While Toil encodes paths into a URL with ToilPathMapper, + # something called internally in cwltool may return an absolute path + # ex: if cwltool calls itself internally in command_line_tool.py, + # it collects outputs with collect_output, and revmap_file will use its own internal pathmapper + resolved = location + else: + raise RuntimeError("Unsupported location: " + location) + + if location in self._pathmap: + # Don't map the same directory twice + logger.debug( + "ToilPathMapper stopping recursion because we have already " + "mapped directory: %s", + location, + ) + return + + logger.debug( + "ToilPathMapper adding directory mapping %s -> %s", resolved, tgt + ) + self._pathmap[location] = MapperEnt( + resolved, + tgt, + "WritableDirectory" if (copy or copy_here) else "Directory", + staged, + ) + + if not location.startswith("_:") and not self.stage_listing: + # Don't stage anything below here separately, since we are able + # to copy the whole directory from somewhere and and we can't + # stage files over themselves. + staged = False + + # Keep recursing + self.visitlisting( + obj.get("listing", []), + tgt, + basedir, + copy=copy, + staged=staged, + ) + + if is_file(obj): + logger.debug("ToilPathMapper visiting file %s", location) + + if location in self._pathmap: + # Don't map the same file twice + logger.debug( + "ToilPathMapper stopping recursion because we have already " + "mapped file: %s", + location, + ) + return + + ab = abspath(location, basedir) + if "contents" in obj and location.startswith("_:"): + # We are supposed to create this file + self._pathmap[location] = MapperEnt( + obj["contents"], + tgt, + "CreateWritableFile" if copy else "CreateFile", + staged, + ) + else: + with SourceLine( + obj, + "location", + ValidationException, + logger.isEnabledFor(logging.DEBUG), + ): + # If we have access to the Toil file store, we will have a + # get_file set, and it will convert this path to a file: + # URI for a local file it downloaded. + if self.get_file: + deref = self.get_file( + location, + obj.get("streamable", False), + self.streaming_allowed, + ) + else: + deref = ab + if deref.startswith("file:"): + deref = schema_salad.ref_resolver.uri_file_path(deref) + if urlsplit(deref).scheme in ["http", "https"]: + deref = downloadHttpFile(location) + elif urlsplit(deref).scheme != "toilfile": + # Dereference symbolic links + st = os.lstat(deref) + while stat.S_ISLNK(st.st_mode): + logger.debug("ToilPathMapper following symlink %s", deref) + rl = os.readlink(deref) + deref = ( + rl + if os.path.isabs(rl) + else os.path.join(os.path.dirname(deref), rl) + ) + st = os.lstat(deref) -class ToilExpressionTool(ToilTool, cwltool.command_line_tool.ExpressionTool): - """Subclass the cwltool expression tool to provide the custom ToilPathMapper.""" + # If we didn't download something that is a toilfile: + # reference, we just pass that along. + """Link or copy files to their targets. Create them as needed.""" -def toil_make_tool( - toolpath_object: CommentedMap, - loadingContext: cwltool.context.LoadingContext, -) -> Process: - """ - Emit custom ToilCommandLineTools. + logger.debug( + "ToilPathMapper adding file mapping %s -> %s", deref, tgt + ) - This factory function is meant to be passed to cwltool.load_tool(). - """ - if isinstance(toolpath_object, Mapping): - if toolpath_object.get("class") == "CommandLineTool": - return ToilCommandLineTool(toolpath_object, loadingContext) - elif toolpath_object.get("class") == "ExpressionTool": - return ToilExpressionTool(toolpath_object, loadingContext) - return cwltool.workflow.default_make_tool(toolpath_object, loadingContext) + self._pathmap[location] = MapperEnt( + deref, tgt, "WritableFile" if copy else "File", staged + ) + # Handle all secondary files that need to be next to this one. + self.visitlisting( + obj.get("secondaryFiles", []), + stagedir, + basedir, + copy=copy, + staged=staged, + ) # When a file we want to have is missing, we can give it this sentinel location # URI instead of raising an error right away, in case it is optional. @@ -2345,90 +2445,6 @@ def remove_empty_listings(rec: CWLDirectoryType) -> None: return -class CWLNamedJob(Job): - """ - Base class for all CWL jobs that do user work, to give them useful names. - """ - - def __init__( - self, - cores: float | None = 1, - memory: int | str | None = "1GiB", - disk: int | str | None = "1MiB", - accelerators: list[AcceleratorRequirement] | None = None, - preemptible: bool | None = None, - walltime: int | None = 0, - tool_id: str | None = None, - parent_name: str | None = None, - subjob_name: str | None = None, - local: bool | None = None, - ) -> None: - """ - Make a new job and set up its requirements and naming. - - :param tool_id: Full CWL tool ID for the job, if applicable. - :param parent_name: Shortened name of the parent CWL job, if applicable. - :param subjob_name: Name qualifier for when one CWL tool becomes multiple jobs. - """ - - # Get the name of the class we are, as a final fallback or a name - # component. - class_name = self.__class__.__name__ - - name_parts = [] - - if parent_name is not None: - # Scope to parent - name_parts.append(parent_name) - if tool_id is not None: - # Start with the short name of the tool - name_parts.append(shortname(tool_id)) - if subjob_name is not None: - # We aren't this whole thing, we're a sub-component - name_parts.append(subjob_name) - if tool_id is None and subjob_name is None: - # We need something. Put the class. - name_parts.append(class_name) - - # Dotted hierarchical name used both as the unit name and as - # file hints for writes to the job store. - self.task_path = ".".join(name_parts) - - # Display as that along with the class - display_name = f"{class_name} {self.task_path}" - - # Set up the job with the right requirements and names. - super().__init__( - cores=cores, - memory=memory, - disk=disk, - accelerators=accelerators, - preemptible=preemptible, - walltime=walltime, - unitName=self.task_path, - displayName=display_name, - local=local, - ) - - -class ResolveIndirect(CWLNamedJob): - """ - Helper Job. - - Accepts an unresolved dict (containing promises) and produces a dictionary - of actual values. - """ - - def __init__(self, cwljob: Promised[CWLObjectType], parent_name: str | None = None): - """Store the dictionary of promises for later resolution.""" - super().__init__(parent_name=parent_name, subjob_name="_resolve", local=True) - self.cwljob = cwljob - - def run(self, file_store: AbstractFileStore) -> CWLObjectType: - """Evaluate the promises and return their values.""" - return resolve_dict_w_promises(unwrap(self.cwljob)) - - def toilStageFiles( toil: Toil, cwljob: CWLObjectType | list[CWLObjectType], @@ -2618,6 +2634,101 @@ def _check_adjust( visit_class(cwljob, ("File", "Directory"), _check_adjust) +##### +# +# In this section, we define the Toil Job classes used to build the Toil job +# graph that runs a CWL workflow. +# +# We have a base CWLNamedJob, and from that we get a CWLJobWrapper and CWLJob +# for actually executing steps, and some other jobs to glue those together into +# workflows with the right control flow structures. +# +### + +class CWLNamedJob(Job): + """ + Base class for all CWL jobs that do user work, to give them useful names. + """ + + def __init__( + self, + cores: float | None = 1, + memory: int | str | None = "1GiB", + disk: int | str | None = "1MiB", + accelerators: list[AcceleratorRequirement] | None = None, + preemptible: bool | None = None, + walltime: int | None = 0, + tool_id: str | None = None, + parent_name: str | None = None, + subjob_name: str | None = None, + local: bool | None = None, + ) -> None: + """ + Make a new job and set up its requirements and naming. + + :param tool_id: Full CWL tool ID for the job, if applicable. + :param parent_name: Shortened name of the parent CWL job, if applicable. + :param subjob_name: Name qualifier for when one CWL tool becomes multiple jobs. + """ + + # Get the name of the class we are, as a final fallback or a name + # component. + class_name = self.__class__.__name__ + + name_parts = [] + + if parent_name is not None: + # Scope to parent + name_parts.append(parent_name) + if tool_id is not None: + # Start with the short name of the tool + name_parts.append(shortname(tool_id)) + if subjob_name is not None: + # We aren't this whole thing, we're a sub-component + name_parts.append(subjob_name) + if tool_id is None and subjob_name is None: + # We need something. Put the class. + name_parts.append(class_name) + + # Dotted hierarchical name used both as the unit name and as + # file hints for writes to the job store. + self.task_path = ".".join(name_parts) + + # Display as that along with the class + display_name = f"{class_name} {self.task_path}" + + # Set up the job with the right requirements and names. + super().__init__( + cores=cores, + memory=memory, + disk=disk, + accelerators=accelerators, + preemptible=preemptible, + walltime=walltime, + unitName=self.task_path, + displayName=display_name, + local=local, + ) + + +class ResolveIndirect(CWLNamedJob): + """ + Helper Job. + + Accepts an unresolved dict (containing promises) and produces a dictionary + of actual values. + """ + + def __init__(self, cwljob: Promised[CWLObjectType], parent_name: str | None = None): + """Store the dictionary of promises for later resolution.""" + super().__init__(parent_name=parent_name, subjob_name="_resolve", local=True) + self.cwljob = cwljob + + def run(self, file_store: AbstractFileStore) -> CWLObjectType: + """Evaluate the promises and return their values.""" + return resolve_dict_w_promises(unwrap(self.cwljob)) + + class CWLJobWrapper(CWLNamedJob): """ Wrap a CWL job that uses dynamic resources requirement. @@ -3089,14 +3200,6 @@ def file_import_function(url: str, log_level: int = logging.DEBUG) -> FileID: return output -def get_container_engine(runtime_context: cwltool.context.RuntimeContext) -> str: - if runtime_context.podman: - return "podman" - elif runtime_context.singularity: - return "singularity" - return "docker" - - def makeRootJob( tool: Process, jobobj: CWLObjectType, @@ -3565,9 +3668,9 @@ def __init__( :param parent_name: human-readable name prefix used for child job naming :param iteration_limit: maximum number of iterations before raising an error :param iteration: zero-based index of the current iteration we would execute - :param previous_outputs: promise for the previous iteration's output dict; + :param previous_outputs: promise for the previous iteration's output dict; should be None for iteration 0 - :param previous_accumulation: accumulated outputs from all prior iterations; + :param previous_accumulation: accumulated outputs from all prior iterations; used for outputMethod: all_iterations; should be None for iteration 0 """ super().__init__(cores=1, memory=INTERPRETER_JOB_MEMORY, disk=INTERPRETER_JOB_DISK, local=True) @@ -3579,7 +3682,7 @@ def __init__( self.iteration = iteration self.previous_outputs = previous_outputs self.previous_accumulation = previous_accumulation - + def build_next_inputs(self, cwljob: CWLObjectType, body_followon: Job, loop_inputs: list[CWLObjectType]) -> UnresolvedDict: """ Build the input object for the next iteration by applying the @@ -3650,7 +3753,7 @@ def build_next_inputs(self, cwljob: CWLObjectType, body_followon: Job, loop_inpu # Note: the step's requirements have already included the workflow's requirements # https://github.com/common-workflow-language/cwltool/blob/1bf74499ca1c4a5f98e7cffb0ad4aa89aa98cb9e/cwltool/workflow.py#L207 if not bool(self.step.get_requirement("StepInputExpressionRequirement")[0]): - # Raise the same error cwltool does: + # Raise the same error cwltool does: # https://github.com/common-workflow-language/cwltool/blob/1bf74499ca1c4a5f98e7cffb0ad4aa89aa98cb9e/cwltool/workflow_job.py#L622 raise cwl_utils.errors.WorkflowException( "Workflow step contains valueFrom but StepInputExpressionRequirement not in requirements" @@ -3738,7 +3841,7 @@ def run(self, file_store: AbstractFileStore) -> CWLObjectType | Promised[CWLObje ) self.addChild(body_job) - + # Declare before the if/else so that mypy can see that they are always defined next_loop_pred: CWLLoopAccumulate | Job next_accumulation: Promised[CWLObjectType] | None @@ -3765,7 +3868,7 @@ def run(self, file_store: AbstractFileStore) -> CWLObjectType | Promised[CWLObje # updated accumulator as its predecessor. next_loop_pred = accumulated_loops next_accumulation = accumulated_loops.rv() - + # In last_iteration mode we don't need to accumulate anything - # the next CWLLoop runs directly after the body job, and # next_accumulation stays None since we only care about the final @@ -3773,7 +3876,7 @@ def run(self, file_store: AbstractFileStore) -> CWLObjectType | Promised[CWLObje elif output_method == "last_iteration": next_loop_pred = body_followon next_accumulation = None - + else: raise cwl_utils.errors.WorkflowException( f"Unsupported cwltool:Loop outputMethod {output_method!r}" @@ -3791,10 +3894,10 @@ def run(self, file_store: AbstractFileStore) -> CWLObjectType | Promised[CWLObje previous_outputs=body_followon.rv(), previous_accumulation=next_accumulation, ) - + # Attach next loop to its predecessor, which depends on the output method next_loop_pred.addFollowOn(next_loop) - + return next_loop.rv() @@ -3960,7 +4063,7 @@ def run( requirements=self.cwlwf.requirements, container_engine=get_container_engine(self.runtime_context), ) - + # Declare types for mypy so it can see that they are always defined in the if/else below wfjob: CWLLoop | CWLScatter | CWLWorkflow | CWLJob | CWLJobWrapper followOn: CWLLoop | CWLGather | ResolveIndirect | CWLJob | CWLJobWrapper @@ -4318,6 +4421,12 @@ def run(self, file_store: AbstractFileStore) -> Any: self.addChild(cwljob) return cwljob.rv() +##### +# +# In this section, we have helper functions for loading and setting uop a +# workflow run, and the main CLI entry point function to run a CWL file. +# +##### def extract_workflow_inputs( options: Namespace, initialized_job_order: CWLObjectType, tool: Process @@ -4673,6 +4782,17 @@ def determine_load_listing( return cast(Literal["no_listing", "shallow_listing", "deep_listing"], load_listing) +def find_default_container( + args: Namespace, builder: cwltool.builder.Builder +) -> str | None: + """Find the default constructor by consulting a Toil.options object.""" + if args.default_container: + return str(args.default_container) + if args.beta_use_biocontainers: + return get_container_from_software_requirements(True, builder) + return None + + usage_message = "\n\n" + textwrap.dedent(""" NOTE: If you're trying to specify a jobstore, you must use --jobStore, not a positional argument. @@ -5149,14 +5269,3 @@ def remove_at_id(doc: Any) -> None: return 1 return 0 - - -def find_default_container( - args: Namespace, builder: cwltool.builder.Builder -) -> str | None: - """Find the default constructor by consulting a Toil.options object.""" - if args.default_container: - return str(args.default_container) - if args.beta_use_biocontainers: - return get_container_from_software_requirements(True, builder) - return None diff --git a/src/toil/test/cwl/fail_with_log.cwl b/src/toil/test/cwl/fail_with_log.cwl index f5277b80c4..c9fc9e2a85 100644 --- a/src/toil/test/cwl/fail_with_log.cwl +++ b/src/toil/test/cwl/fail_with_log.cwl @@ -15,7 +15,7 @@ steps: baseCommand: [bash] arguments: [ "-c", "echo >&2 '$(inputs.s)' ; exit 1" ] stdout: output_log.txt - stderr: error_log.txt + stderr: error_log_secret.txt outputs: out_file: type: File From 9fe170545ed701d21b0c171d4a4cc2a5d5563cd4 Mon Sep 17 00:00:00 2001 From: Adam Novak Date: Thu, 16 Jul 2026 15:19:43 -0400 Subject: [PATCH 3/5] Hook the job method of all ToilTool Processes in a way we can sneak past MyPy --- src/toil/cwl/cwltoil.py | 172 ++++++++++++++++++++++++++++++++++++++-- 1 file changed, 165 insertions(+), 7 deletions(-) diff --git a/src/toil/cwl/cwltoil.py b/src/toil/cwl/cwltoil.py index 299be73c2f..257ea1b19d 100644 --- a/src/toil/cwl/cwltoil.py +++ b/src/toil/cwl/cwltoil.py @@ -16,8 +16,13 @@ # See the License for the specific language governing permissions and # limitations under the License. -# For an overview of how this all works, see discussion in -# docs/architecture.rst + +##### +# For an overview of how this all works, see the section "Toil support for +# Common Workflow Language" in docs/appendicies/architecture.rst +##### + +import abc import base64 import copy import datetime @@ -48,7 +53,18 @@ ) from tempfile import NamedTemporaryFile, TemporaryFile, gettempdir from threading import Thread -from typing import IO, Any, Literal, Optional, Protocol, TextIO, TypeVar, Union, cast +from typing import ( + IO, + Any, + Generic, + Literal, + Optional, + Protocol, + TextIO, + TypeVar, + Union, + cast +) from urllib.parse import quote, unquote, urlparse, urlsplit import cwl_utils.errors @@ -830,8 +846,45 @@ def toil_make_tool( # This goes on to install our PathMapper hooks. ##### -class ToilTool(Process): - """Mixin to hook Toil into a cwltool tool type.""" +# Processes in general can generate a lot of types of jobs, but CommandLineTools and ExpressionTools are typed as generating a more restricted set of possible types of jobs. +# We want to be able to wrap them with the same wrapper. +# +# ExperssionTool's job() is also able to take a None for output_callbacks, while CommandLineTool's can't. +# +# We can't just say the wrapper mixin generates the more permissive type range because then it won't be substitutable for a CommandLineTool, etc. +# +# So to type this we have to do a bunch of ugly type stuff, and event hen we need to cast (see https://github.com/python/mypy/issues/17192#issuecomment-4995017670) + +JobTypesT = TypeVar("JobTypesT", covariant=True) +CallbackTypesT = TypeVar( + "CallbackTypesT", + bound=OutputCallbackType | None, + contravariant=True +) +class ProcessProtocol(Protocol, Generic[JobTypesT, CallbackTypesT]): + """ + Protocol for the type of a cwltool Process that yields jobs of the given + union type and accepts output_callbacks of the givne union type on its + run() method. + """ + def job( + self, + job_order: CWLObjectType, + output_callbacks: CallbackTypesT, + runtimeContext: cwltool.context.RuntimeContext, + ) -> Generator[JobTypesT, None, None]: + ... + +class ToilTool(Generic[JobTypesT, CallbackTypesT]): + """ + Mixin to hook Toil into a cwltool tool type. + + This MUST be mixed in earlier in the method resolution order than an actual + concrete implementation of cwltool.process.Process. + + Ypu also need to tell it the types that the mixed-with Process uses on its + run() method, so it can be wrapped with the right types. + """ def __init__(self, *args: Any, **kwargs: Any) -> None: """ @@ -840,6 +893,8 @@ def __init__(self, *args: Any, **kwargs: Any) -> None: super().__init__(*args, **kwargs) # Reserve a spot for the Toil job that ends up executing this tool. self._toil_job: Job | None = None + # Reserve a spot for the file store we are executing against. + self._toil_file_store: AbstractFileStore | None = None # Remember path mappers we have used so we can interrogate them later to find out what the job mapped. self._path_mappers: list[cwltool.pathmapper.PathMapper] = [] @@ -848,9 +903,95 @@ def connect_toil_job(self, job: Job) -> None: Attach the Toil tool to the Toil job that is executing it. This allows it to use the Toil job to stop at certain points if debugging flags are set. + + Must be called before the tool is executed. """ self._toil_job = job + def connect_toil_file_store(self, file_store: AbstractFileStore) -> None: + """ + Attach the Toil tool to the Toil file store it is running against. + + This is needed in order to allow hooks we add into the cwltool + tool-running code to talk to the file stroe for things like logging. + + Must be called before the tool is executed. + """ + self._toil_file_store = file_store + + def job( + self, + job_order: CWLObjectType, + output_callbacks: CallbackTypesT, + runtimeContext: cwltool.context.RuntimeContext, + ) -> Generator[JobTypesT, None, None]: + """ + Attach Toil postprocessing hooks that should apply to all CWL jobs. + + To get a chance to run code after all the important stuff about the job + (temp directory, stderr path, etc.) has been determined, but before it + all gets deleted by cwltool's cleanup logic, we add "output callbacks". + """ + + # We want to override job() and then call the job() of the Process we + # are mixed into. MyPy has no good way to tell it that there will be a + # job() method later in the method resolution order than we are, when + # the mixin is used correctly. See + # . So we just tell it + # very firmly that things will be OK and our super() actually does have + # the method we need. + # + # TODO: Can we avoid this somehow? + + s = cast(ProcessProtocol[JobTypesT, CallbackTypesT], super()) + for job in s.job(job_order, output_callbacks, runtimeContext): + + # If we didn't redirect standard error in the .cwl tool definition, + # we can collect it from the default location we pass into the + # executor. If we did redirect it, and the job fails, the user will + # want it, but they won't be able to get it easily by just looking + # in the right directory; we have to ship it to the leader before + # it gets cleaned up. + + if isinstance(job, cwltool.job.JobBase): + + def log_stderr_output_callback( + output_callbacks: OutputCallbackType | None, + outputs: CWLObjectType | None, + processStatus: str, + ) -> None: + """ + Output callback that logs failed job standard error. + + Needs to be partial'd into a callback chain. + """ + + # In the callback, we have the various job fields about + # error logging filled in. + + # This comes from JobBase._setup() + assert isinstance(job.base_path_logs, str) + + # Make sure the Toil job running us did the right connect + # calls, so the type checker knows. + assert self._toil_job is not None + assert self._toil_file_store is not None + + if processStatus != "success" and job.stderr is not None: + # We know this is where cwltool puts standard error + # when you send it to a file. + stderr_path = os.path.join(job.base_path_logs, job.stderr) + self._toil_file_store.log_user_stream( + self._toil_job.description.unitName + ".stderr", + open(stderr_path, 'rb') + ) + + if output_callbacks: + output_callbacks(outputs, processStatus) + + job.output_callback = functools.partial(log_stderr_output_callback, job.output_callback) + + yield job def make_path_mapper( self, @@ -884,7 +1025,15 @@ def __str__(self) -> str: """Return string representation of this tool type.""" return f'{self.__class__.__name__}({repr(getattr(self, "tool", {}).get("id", "???"))})' -class ToilCommandLineTool(ToilTool, cwltool.command_line_tool.CommandLineTool): +# TODO: Is there a way we can make the type we're mixing with the generic +# parameter and pull out the return type of its run method? +class ToilCommandLineTool( + ToilTool[ + cwltool.job.JobBase | cwltool.command_line_tool.CallbackJob, + OutputCallbackType, + ], + cwltool.command_line_tool.CommandLineTool, +): """ CWL CommandLineTool that connects to ToilPathMapper and other hooks. """ @@ -948,9 +1097,16 @@ def _initialworkdir( self._toil_job.files_downloaded_hook(host_and_job_paths) -class ToilExpressionTool(ToilTool, cwltool.command_line_tool.ExpressionTool): +class ToilExpressionTool( + ToilTool[ + cwltool.command_line_tool.ExpressionJob, + OutputCallbackType | None, + ], + cwltool.command_line_tool.ExpressionTool, +): """Subclass the cwltool expression tool to provide the custom ToilPathMapper.""" # TODO: Don't expression tools also want the resource monitoring hooks? + pass ##### # Our other entry point into cwltool for customizing running things is by @@ -3115,6 +3271,8 @@ def run(self, file_store: AbstractFileStore) -> Any: # Connect the CWL tool to us so it can call into the Toil job when # it reaches points where we might need to debug it. self.cwltool.connect_toil_job(self) + # And to the file store so it can do logging to the leader + self.cwltool.connect_toil_file_store(file_store) status = "did_not_run" try: From 26888a1db15f61c8d033733ea88e9f2c7c98f06a Mon Sep 17 00:00:00 2001 From: Adam Novak Date: Thu, 16 Jul 2026 17:05:57 -0400 Subject: [PATCH 4/5] Add test to make sure log is returned. --- src/toil/test/cwl/cwlTest.py | 20 ++++++++++++++++++++ src/toil/test/cwl/fail_with_log.cwl | 4 ++-- 2 files changed, 22 insertions(+), 2 deletions(-) diff --git a/src/toil/test/cwl/cwlTest.py b/src/toil/test/cwl/cwlTest.py index 7d79ec5278..a61a22284a 100644 --- a/src/toil/test/cwl/cwlTest.py +++ b/src/toil/test/cwl/cwlTest.py @@ -1884,6 +1884,8 @@ def test_wes_server_cwl_conformance(self, cwl_v1_2_spec: Path) -> None: ) +# TODO: Why aren't these in TestCWLWorkflow? They run workflows. + @needs_cwl @pytest.mark.cwl @pytest.mark.cwl_small_log_dir @@ -2128,6 +2130,24 @@ def test_workflow_echo_string_scatter_capture_stdout(tmp_path: Path) -> None: assert "Finished toil run successfully" in p.stderr assert p.returncode == 0 +@needs_cwl +@pytest.mark.cwl +@pytest.mark.cwl_small +def test_captured_stderr_for_failed_job_is_reported(tmp_path: Path) -> None: + """ + Make sure that standard error for failed jobs is logged, even when the job + is capturing standard error. + """ + with get_data("test/cwl/fail_with_log.cwl") as cwl_file: + cmd = [ + "toil-cwl-runner", + f"--jobStore=file:{tmp_path / 'jobStore'}", + str(cwl_file), + ] + p = subprocess.run(cmd, capture_output=True, text=True) + assert p.returncode != 0 + assert "Message: This is a test" in p.stderr + @needs_cwl @pytest.mark.cwl diff --git a/src/toil/test/cwl/fail_with_log.cwl b/src/toil/test/cwl/fail_with_log.cwl index c9fc9e2a85..138792c640 100644 --- a/src/toil/test/cwl/fail_with_log.cwl +++ b/src/toil/test/cwl/fail_with_log.cwl @@ -13,9 +13,9 @@ steps: inputs: s: string baseCommand: [bash] - arguments: [ "-c", "echo >&2 '$(inputs.s)' ; exit 1" ] + arguments: [ "-c", "printf >&2 'Message: '; echo >&2 '$(inputs.s)' ; exit 1" ] stdout: output_log.txt - stderr: error_log_secret.txt + stderr: error_log.txt outputs: out_file: type: File From 7b06592b092c22001b9d64ebb69fb9668f4ef9e6 Mon Sep 17 00:00:00 2001 From: Adam Novak Date: Thu, 16 Jul 2026 17:14:52 -0400 Subject: [PATCH 5/5] Add code in case something goes really wrong and the log is not there --- src/toil/cwl/cwltoil.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/src/toil/cwl/cwltoil.py b/src/toil/cwl/cwltoil.py index 257ea1b19d..22db8dab8a 100644 --- a/src/toil/cwl/cwltoil.py +++ b/src/toil/cwl/cwltoil.py @@ -981,10 +981,13 @@ def log_stderr_output_callback( # We know this is where cwltool puts standard error # when you send it to a file. stderr_path = os.path.join(job.base_path_logs, job.stderr) - self._toil_file_store.log_user_stream( - self._toil_job.description.unitName + ".stderr", - open(stderr_path, 'rb') - ) + if os.path.isfile(stderr_path): + self._toil_file_store.log_user_stream( + self._toil_job.description.unitName + ".stderr", + open(stderr_path, 'rb') + ) + else: + logger.error("CWL job standard error at %s is missing!", stderr_path) if output_callbacks: output_callbacks(outputs, processStatus)