--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/env/lib/python3.9/site-packages/cwltool/process.py	Mon Mar 22 18:12:50 2021 +0000
@@ -0,0 +1,1298 @@
+"""Classes and methods relevant for all CWL Proccess types."""
+import abc
+import copy
+import functools
+import hashlib
+import json
+import logging
+import math
+import os
+import shutil
+import stat
+import textwrap
+import urllib
+import uuid
+from os import scandir
+from typing import (
+    Any,
+    Callable,
+    Dict,
+    Iterable,
+    Iterator,
+    List,
+    MutableMapping,
+    MutableSequence,
+    Optional,
+    Set,
+    Sized,
+    Tuple,
+    Type,
+    Union,
+    cast,
+)
+
+from pkg_resources import resource_stream
+from rdflib import Graph
+from ruamel.yaml.comments import CommentedMap, CommentedSeq
+from schema_salad.avro.schema import (
+    Names,
+    Schema,
+    SchemaParseException,
+    make_avsc_object,
+)
+from schema_salad.exceptions import ValidationException
+from schema_salad.ref_resolver import Loader, file_uri, uri_file_path
+from schema_salad.schema import load_schema, make_avro_schema, make_valid_avro
+from schema_salad.sourceline import SourceLine, strip_dup_lineno
+from schema_salad.utils import convert_to_dict
+from schema_salad.validate import validate_ex
+from typing_extensions import TYPE_CHECKING
+
+from . import expression
+from .builder import Builder, HasReqsHints
+from .context import LoadingContext, RuntimeContext, getdefault
+from .errors import UnsupportedRequirement, WorkflowException
+from .loghandler import _logger
+from .mpi import MPIRequirementName
+from .pathmapper import MapperEnt, PathMapper
+from .secrets import SecretStore
+from .stdfsaccess import StdFsAccess
+from .update import INTERNAL_VERSION
+from .utils import (
+    CWLObjectType,
+    CWLOutputAtomType,
+    CWLOutputType,
+    JobsGeneratorType,
+    OutputCallbackType,
+    adjustDirObjs,
+    aslist,
+    cmp_like_py2,
+    copytree_with_merge,
+    ensure_writable,
+    get_listing,
+    normalizeFilesDirs,
+    onWindows,
+    random_outdir,
+    visit_class,
+)
+from .validate_js import validate_js_expressions
+
+if TYPE_CHECKING:
+    from .provenance_profile import ProvenanceProfile  # pylint: disable=unused-import
+
+
+class LogAsDebugFilter(logging.Filter):
+    def __init__(self, name: str, parent: logging.Logger) -> None:
+        """Initialize."""
+        name = str(name)
+        super().__init__(name)
+        self.parent = parent
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        return self.parent.isEnabledFor(logging.DEBUG)
+
+
+_logger_validation_warnings = logging.getLogger("cwltool.validation_warnings")
+_logger_validation_warnings.setLevel(_logger.getEffectiveLevel())
+_logger_validation_warnings.addFilter(
+    LogAsDebugFilter("cwltool.validation_warnings", _logger)
+)
+
+supportedProcessRequirements = [
+    "DockerRequirement",
+    "SchemaDefRequirement",
+    "EnvVarRequirement",
+    "ScatterFeatureRequirement",
+    "SubworkflowFeatureRequirement",
+    "MultipleInputFeatureRequirement",
+    "InlineJavascriptRequirement",
+    "ShellCommandRequirement",
+    "StepInputExpressionRequirement",
+    "ResourceRequirement",
+    "InitialWorkDirRequirement",
+    "ToolTimeLimit",
+    "WorkReuse",
+    "NetworkAccess",
+    "InplaceUpdateRequirement",
+    "LoadListingRequirement",
+    MPIRequirementName,
+    "http://commonwl.org/cwltool#TimeLimit",
+    "http://commonwl.org/cwltool#WorkReuse",
+    "http://commonwl.org/cwltool#NetworkAccess",
+    "http://commonwl.org/cwltool#LoadListingRequirement",
+    "http://commonwl.org/cwltool#InplaceUpdateRequirement",
+]
+
+cwl_files = (
+    "Workflow.yml",
+    "CommandLineTool.yml",
+    "CommonWorkflowLanguage.yml",
+    "Process.yml",
+    "Operation.yml",
+    "concepts.md",
+    "contrib.md",
+    "intro.md",
+    "invocation.md",
+)
+
+salad_files = (
+    "metaschema.yml",
+    "metaschema_base.yml",
+    "salad.md",
+    "field_name.yml",
+    "import_include.md",
+    "link_res.yml",
+    "ident_res.yml",
+    "vocab_res.yml",
+    "vocab_res.yml",
+    "field_name_schema.yml",
+    "field_name_src.yml",
+    "field_name_proc.yml",
+    "ident_res_schema.yml",
+    "ident_res_src.yml",
+    "ident_res_proc.yml",
+    "link_res_schema.yml",
+    "link_res_src.yml",
+    "link_res_proc.yml",
+    "vocab_res_schema.yml",
+    "vocab_res_src.yml",
+    "vocab_res_proc.yml",
+)
+
+SCHEMA_CACHE = (
+    {}
+)  # type: Dict[str, Tuple[Loader, Union[Names, SchemaParseException], CWLObjectType, Loader]]
+SCHEMA_FILE = None  # type: Optional[CWLObjectType]
+SCHEMA_DIR = None  # type: Optional[CWLObjectType]
+SCHEMA_ANY = None  # type: Optional[CWLObjectType]
+
+custom_schemas = {}  # type: Dict[str, Tuple[str, str]]
+
+
+def use_standard_schema(version: str) -> None:
+    if version in custom_schemas:
+        del custom_schemas[version]
+    if version in SCHEMA_CACHE:
+        del SCHEMA_CACHE[version]
+
+
+def use_custom_schema(version: str, name: str, text: str) -> None:
+    custom_schemas[version] = (name, text)
+    if version in SCHEMA_CACHE:
+        del SCHEMA_CACHE[version]
+
+
+def get_schema(
+    version: str,
+) -> Tuple[Loader, Union[Names, SchemaParseException], CWLObjectType, Loader]:
+
+    if version in SCHEMA_CACHE:
+        return SCHEMA_CACHE[version]
+
+    cache = {}  # type: Dict[str, Union[str, Graph, bool]]
+    version = version.split("#")[-1]
+    if ".dev" in version:
+        version = ".".join(version.split(".")[:-1])
+    for f in cwl_files:
+        try:
+            res = resource_stream(__name__, f"schemas/{version}/{f}")
+            cache["https://w3id.org/cwl/" + f] = res.read().decode("UTF-8")
+            res.close()
+        except OSError:
+            pass
+
+    for f in salad_files:
+        try:
+            res = resource_stream(
+                __name__,
+                f"schemas/{version}/salad/schema_salad/metaschema/{f}",
+            )
+            cache[
+                "https://w3id.org/cwl/salad/schema_salad/metaschema/" + f
+            ] = res.read().decode("UTF-8")
+            res.close()
+        except OSError:
+            pass
+
+    if version in custom_schemas:
+        cache[custom_schemas[version][0]] = custom_schemas[version][1]
+        SCHEMA_CACHE[version] = load_schema(custom_schemas[version][0], cache=cache)
+    else:
+        SCHEMA_CACHE[version] = load_schema(
+            "https://w3id.org/cwl/CommonWorkflowLanguage.yml", cache=cache
+        )
+
+    return SCHEMA_CACHE[version]
+
+
+def shortname(inputid: str) -> str:
+    d = urllib.parse.urlparse(inputid)
+    if d.fragment:
+        return d.fragment.split("/")[-1]
+    return d.path.split("/")[-1]
+
+
+def checkRequirements(
+    rec: Union[MutableSequence[CWLObjectType], CWLObjectType, CWLOutputType, None],
+    supported_process_requirements: Iterable[str],
+) -> None:
+    if isinstance(rec, MutableMapping):
+        if "requirements" in rec:
+            for i, entry in enumerate(
+                cast(MutableSequence[CWLObjectType], rec["requirements"])
+            ):
+                with SourceLine(rec["requirements"], i, UnsupportedRequirement):
+                    if cast(str, entry["class"]) not in supported_process_requirements:
+                        raise UnsupportedRequirement(
+                            "Unsupported requirement {}".format(entry["class"])
+                        )
+        for key in rec:
+            checkRequirements(rec[key], supported_process_requirements)
+    if isinstance(rec, MutableSequence):
+        for entry2 in rec:
+            checkRequirements(entry2, supported_process_requirements)
+
+
+def stage_files(
+    pathmapper: PathMapper,
+    stage_func: Optional[Callable[[str, str], None]] = None,
+    ignore_writable: bool = False,
+    symlink: bool = True,
+    secret_store: Optional[SecretStore] = None,
+    fix_conflicts: bool = False,
+) -> None:
+    """Link or copy files to their targets. Create them as needed."""
+    targets = {}  # type: Dict[str, MapperEnt]
+    for key, entry in pathmapper.items():
+        if "File" not in entry.type:
+            continue
+        if entry.target not in targets:
+            targets[entry.target] = entry
+        elif targets[entry.target].resolved != entry.resolved:
+            if fix_conflicts:
+                # find first key that does not clash with an existing entry in targets
+                # start with entry.target + '_' + 2 and then keep incrementing the number till there is no clash
+                i = 2
+                tgt = f"{entry.target}_{i}"
+                while tgt in targets:
+                    i += 1
+                    tgt = f"{entry.target}_{i}"
+                targets[tgt] = pathmapper.update(
+                    key, entry.resolved, tgt, entry.type, entry.staged
+                )
+            else:
+                raise WorkflowException(
+                    "File staging conflict, trying to stage both %s and %s to the same target %s"
+                    % (targets[entry.target].resolved, entry.resolved, entry.target)
+                )
+
+    for key, entry in pathmapper.items():
+        if not entry.staged:
+            continue
+        if not os.path.exists(os.path.dirname(entry.target)):
+            os.makedirs(os.path.dirname(entry.target))
+        if entry.type in ("File", "Directory") and os.path.exists(entry.resolved):
+            if symlink:  # Use symlink func if allowed
+                if onWindows():
+                    if entry.type == "File":
+                        shutil.copy(entry.resolved, entry.target)
+                    elif entry.type == "Directory":
+                        if os.path.exists(entry.target) and os.path.isdir(entry.target):
+                            shutil.rmtree(entry.target)
+                        copytree_with_merge(entry.resolved, entry.target)
+                else:
+                    os.symlink(entry.resolved, entry.target)
+            elif stage_func is not None:
+                stage_func(entry.resolved, entry.target)
+        elif (
+            entry.type == "Directory"
+            and not os.path.exists(entry.target)
+            and entry.resolved.startswith("_:")
+        ):
+            os.makedirs(entry.target)
+        elif entry.type == "WritableFile" and not ignore_writable:
+            shutil.copy(entry.resolved, entry.target)
+            ensure_writable(entry.target)
+        elif entry.type == "WritableDirectory" and not ignore_writable:
+            if entry.resolved.startswith("_:"):
+                os.makedirs(entry.target)
+            else:
+                shutil.copytree(entry.resolved, entry.target)
+                ensure_writable(entry.target)
+        elif entry.type == "CreateFile" or entry.type == "CreateWritableFile":
+            with open(entry.target, "wb") as new:
+                if secret_store is not None:
+                    new.write(
+                        cast(str, secret_store.retrieve(entry.resolved)).encode("utf-8")
+                    )
+                else:
+                    new.write(entry.resolved.encode("utf-8"))
+            if entry.type == "CreateFile":
+                os.chmod(entry.target, stat.S_IRUSR)  # Read only
+            else:  # it is a "CreateWritableFile"
+                ensure_writable(entry.target)
+            pathmapper.update(key, entry.target, entry.target, entry.type, entry.staged)
+
+
+def relocateOutputs(
+    outputObj: CWLObjectType,
+    destination_path: str,
+    source_directories: Set[str],
+    action: str,
+    fs_access: StdFsAccess,
+    compute_checksum: bool = True,
+    path_mapper: Type[PathMapper] = PathMapper,
+) -> CWLObjectType:
+    adjustDirObjs(outputObj, functools.partial(get_listing, fs_access, recursive=True))
+
+    if action not in ("move", "copy"):
+        return outputObj
+
+    def _collectDirEntries(
+        obj: Union[CWLObjectType, MutableSequence[CWLObjectType], None]
+    ) -> Iterator[CWLObjectType]:
+        if isinstance(obj, dict):
+            if obj.get("class") in ("File", "Directory"):
+                yield obj
+            else:
+                for sub_obj in obj.values():
+                    yield from _collectDirEntries(sub_obj)
+        elif isinstance(obj, MutableSequence):
+            for sub_obj in obj:
+                yield from _collectDirEntries(sub_obj)
+
+    def _relocate(src: str, dst: str) -> None:
+        if src == dst:
+            return
+
+        # If the source is not contained in source_directories we're not allowed to delete it
+        src = fs_access.realpath(src)
+        src_can_deleted = any(
+            os.path.commonprefix([p, src]) == p for p in source_directories
+        )
+
+        _action = "move" if action == "move" and src_can_deleted else "copy"
+
+        if _action == "move":
+            _logger.debug("Moving %s to %s", src, dst)
+            if fs_access.isdir(src) and fs_access.isdir(dst):
+                # merge directories
+                for dir_entry in scandir(src):
+                    _relocate(dir_entry.path, fs_access.join(dst, dir_entry.name))
+            else:
+                shutil.move(src, dst)
+
+        elif _action == "copy":
+            _logger.debug("Copying %s to %s", src, dst)
+            if fs_access.isdir(src):
+                if os.path.isdir(dst):
+                    shutil.rmtree(dst)
+                elif os.path.isfile(dst):
+                    os.unlink(dst)
+                shutil.copytree(src, dst)
+            else:
+                shutil.copy2(src, dst)
+
+    def _realpath(
+        ob: CWLObjectType,
+    ) -> None:  # should be type Union[CWLFile, CWLDirectory]
+        location = cast(str, ob["location"])
+        if location.startswith("file:"):
+            ob["location"] = file_uri(os.path.realpath(uri_file_path(location)))
+        elif location.startswith("/"):
+            ob["location"] = os.path.realpath(location)
+        elif not location.startswith("_:") and ":" in location:
+            ob["location"] = file_uri(fs_access.realpath(location))
+
+    outfiles = list(_collectDirEntries(outputObj))
+    visit_class(outfiles, ("File", "Directory"), _realpath)
+    pm = path_mapper(outfiles, "", destination_path, separateDirs=False)
+    stage_files(pm, stage_func=_relocate, symlink=False, fix_conflicts=True)
+
+    def _check_adjust(a_file: CWLObjectType) -> CWLObjectType:
+        a_file["location"] = file_uri(pm.mapper(cast(str, a_file["location"]))[1])
+        if "contents" in a_file:
+            del a_file["contents"]
+        return a_file
+
+    visit_class(outputObj, ("File", "Directory"), _check_adjust)
+
+    if compute_checksum:
+        visit_class(
+            outputObj, ("File",), functools.partial(compute_checksums, fs_access)
+        )
+    return outputObj
+
+
+def cleanIntermediate(output_dirs: Iterable[str]) -> None:
+    for a in output_dirs:
+        if os.path.exists(a):
+            _logger.debug("Removing intermediate output directory %s", a)
+            shutil.rmtree(a, True)
+
+
+def add_sizes(fsaccess: StdFsAccess, obj: CWLObjectType) -> None:
+    if "location" in obj:
+        try:
+            if "size" not in obj:
+                obj["size"] = fsaccess.size(cast(str, obj["location"]))
+        except OSError:
+            pass
+    elif "contents" in obj:
+        obj["size"] = len(cast(Sized, obj["contents"]))
+    return  # best effort
+
+
+def fill_in_defaults(
+    inputs: List[CWLObjectType],
+    job: CWLObjectType,
+    fsaccess: StdFsAccess,
+) -> None:
+    for e, inp in enumerate(inputs):
+        with SourceLine(
+            inputs, e, WorkflowException, _logger.isEnabledFor(logging.DEBUG)
+        ):
+            fieldname = shortname(cast(str, inp["id"]))
+            if job.get(fieldname) is not None:
+                pass
+            elif job.get(fieldname) is None and "default" in inp:
+                job[fieldname] = copy.deepcopy(inp["default"])
+            elif job.get(fieldname) is None and "null" in aslist(inp["type"]):
+                job[fieldname] = None
+            else:
+                raise WorkflowException(
+                    "Missing required input parameter '%s'"
+                    % shortname(cast(str, inp["id"]))
+                )
+
+
+def avroize_type(
+    field_type: Union[
+        CWLObjectType, MutableSequence[CWLOutputType], CWLOutputType, None
+    ],
+    name_prefix: str = "",
+) -> None:
+    """Add missing information to a type so that CWL types are valid."""
+    if isinstance(field_type, MutableSequence):
+        for field in field_type:
+            avroize_type(field, name_prefix)
+    elif isinstance(field_type, MutableMapping):
+        if field_type["type"] in ("enum", "record"):
+            if "name" not in field_type:
+                field_type["name"] = name_prefix + str(uuid.uuid4())
+        if field_type["type"] == "record":
+            avroize_type(
+                cast(MutableSequence[CWLOutputType], field_type["fields"]), name_prefix
+            )
+        if field_type["type"] == "array":
+            avroize_type(
+                cast(MutableSequence[CWLOutputType], field_type["items"]), name_prefix
+            )
+        if isinstance(field_type["type"], MutableSequence):
+            for ctype in field_type["type"]:
+                avroize_type(cast(CWLOutputType, ctype), name_prefix)
+
+
+def get_overrides(
+    overrides: MutableSequence[CWLObjectType], toolid: str
+) -> CWLObjectType:
+    req = {}  # type: CWLObjectType
+    if not isinstance(overrides, MutableSequence):
+        raise ValidationException(
+            "Expected overrides to be a list, but was %s" % type(overrides)
+        )
+    for ov in overrides:
+        if ov["overrideTarget"] == toolid:
+            req.update(ov)
+    return req
+
+
+_VAR_SPOOL_ERROR = textwrap.dedent(
+    """
+    Non-portable reference to /var/spool/cwl detected: '{}'.
+    To fix, replace /var/spool/cwl with $(runtime.outdir) or add
+    DockerRequirement to the 'requirements' section and declare
+    'dockerOutputDirectory: /var/spool/cwl'.
+    """
+)
+
+
+def var_spool_cwl_detector(
+    obj: CWLOutputType,
+    item: Optional[Any] = None,
+    obj_key: Optional[Any] = None,
+) -> bool:
+    """Detect any textual reference to /var/spool/cwl."""
+    r = False
+    if isinstance(obj, str):
+        if "var/spool/cwl" in obj and obj_key != "dockerOutputDirectory":
+            _logger.warning(
+                SourceLine(item=item, key=obj_key, raise_type=str).makeError(
+                    _VAR_SPOOL_ERROR.format(obj)
+                )
+            )
+            r = True
+    elif isinstance(obj, MutableMapping):
+        for mkey, mvalue in obj.items():
+            r = var_spool_cwl_detector(cast(CWLOutputType, mvalue), obj, mkey) or r
+    elif isinstance(obj, MutableSequence):
+        for lkey, lvalue in enumerate(obj):
+            r = var_spool_cwl_detector(cast(CWLOutputType, lvalue), obj, lkey) or r
+    return r
+
+
+def eval_resource(
+    builder: Builder, resource_req: Union[str, int, float]
+) -> Optional[Union[str, int, float]]:
+    if isinstance(resource_req, str) and expression.needs_parsing(resource_req):
+        result = builder.do_eval(resource_req)
+        if isinstance(result, (str, int)) or result is None:
+            return result
+        raise WorkflowException(
+            "Got incorrect return type {} from resource expression evaluation of {}.".format(
+                type(result), resource_req
+            )
+        )
+    return resource_req
+
+
+# Threshold where the "too many files" warning kicks in
+FILE_COUNT_WARNING = 5000
+
+
+class Process(HasReqsHints, metaclass=abc.ABCMeta):
+    def __init__(
+        self, toolpath_object: CommentedMap, loadingContext: LoadingContext
+    ) -> None:
+        """Build a Process object from the provided dictionary."""
+        super().__init__()
+        self.metadata = getdefault(loadingContext.metadata, {})  # type: CWLObjectType
+        self.provenance_object = None  # type: Optional[ProvenanceProfile]
+        self.parent_wf = None  # type: Optional[ProvenanceProfile]
+        global SCHEMA_FILE, SCHEMA_DIR, SCHEMA_ANY  # pylint: disable=global-statement
+        if SCHEMA_FILE is None or SCHEMA_ANY is None or SCHEMA_DIR is None:
+            get_schema("v1.0")
+            SCHEMA_ANY = cast(
+                CWLObjectType,
+                SCHEMA_CACHE["v1.0"][3].idx["https://w3id.org/cwl/salad#Any"],
+            )
+            SCHEMA_FILE = cast(
+                CWLObjectType,
+                SCHEMA_CACHE["v1.0"][3].idx["https://w3id.org/cwl/cwl#File"],
+            )
+            SCHEMA_DIR = cast(
+                CWLObjectType,
+                SCHEMA_CACHE["v1.0"][3].idx["https://w3id.org/cwl/cwl#Directory"],
+            )
+
+        self.names = make_avro_schema([SCHEMA_FILE, SCHEMA_DIR, SCHEMA_ANY], Loader({}))
+        self.tool = toolpath_object
+        self.requirements = copy.deepcopy(getdefault(loadingContext.requirements, []))
+        self.requirements.extend(self.tool.get("requirements", []))
+        if "id" not in self.tool:
+            self.tool["id"] = "_:" + str(uuid.uuid4())
+        self.requirements.extend(
+            cast(
+                List[CWLObjectType],
+                get_overrides(
+                    getdefault(loadingContext.overrides_list, []), self.tool["id"]
+                ).get("requirements", []),
+            )
+        )
+        self.hints = copy.deepcopy(getdefault(loadingContext.hints, []))
+        self.hints.extend(self.tool.get("hints", []))
+        # Versions of requirements and hints which aren't mutated.
+        self.original_requirements = copy.deepcopy(self.requirements)
+        self.original_hints = copy.deepcopy(self.hints)
+        self.doc_loader = loadingContext.loader
+        self.doc_schema = loadingContext.avsc_names
+
+        self.formatgraph = None  # type: Optional[Graph]
+        if self.doc_loader is not None:
+            self.formatgraph = self.doc_loader.graph
+
+        checkRequirements(self.tool, supportedProcessRequirements)
+        self.validate_hints(
+            cast(Names, loadingContext.avsc_names),
+            self.tool.get("hints", []),
+            strict=getdefault(loadingContext.strict, False),
+        )
+
+        self.schemaDefs = {}  # type: MutableMapping[str, CWLObjectType]
+
+        sd, _ = self.get_requirement("SchemaDefRequirement")
+
+        if sd is not None:
+            sdtypes = cast(MutableSequence[CWLObjectType], sd["types"])
+            avroize_type(cast(MutableSequence[CWLOutputType], sdtypes))
+            av = make_valid_avro(
+                sdtypes,
+                {cast(str, t["name"]): cast(Dict[str, Any], t) for t in sdtypes},
+                set(),
+            )
+            for i in av:
+                self.schemaDefs[i["name"]] = i  # type: ignore
+            make_avsc_object(convert_to_dict(av), self.names)
+
+        # Build record schema from inputs
+        self.inputs_record_schema = {
+            "name": "input_record_schema",
+            "type": "record",
+            "fields": [],
+        }  # type: CWLObjectType
+        self.outputs_record_schema = {
+            "name": "outputs_record_schema",
+            "type": "record",
+            "fields": [],
+        }  # type: CWLObjectType
+
+        for key in ("inputs", "outputs"):
+            for i in self.tool[key]:
+                c = copy.deepcopy(i)
+                c["name"] = shortname(c["id"])
+                del c["id"]
+
+                if "type" not in c:
+                    raise ValidationException(
+                        "Missing 'type' in parameter '{}'".format(c["name"])
+                    )
+
+                if "default" in c and "null" not in aslist(c["type"]):
+                    nullable = ["null"]
+                    nullable.extend(aslist(c["type"]))
+                    c["type"] = nullable
+                else:
+                    c["type"] = c["type"]
+                avroize_type(c["type"], c["name"])
+                if key == "inputs":
+                    cast(
+                        List[CWLObjectType], self.inputs_record_schema["fields"]
+                    ).append(c)
+                elif key == "outputs":
+                    cast(
+                        List[CWLObjectType], self.outputs_record_schema["fields"]
+                    ).append(c)
+
+        with SourceLine(toolpath_object, "inputs", ValidationException):
+            self.inputs_record_schema = cast(
+                CWLObjectType,
+                make_valid_avro(self.inputs_record_schema, {}, set()),
+            )
+            make_avsc_object(convert_to_dict(self.inputs_record_schema), self.names)
+        with SourceLine(toolpath_object, "outputs", ValidationException):
+            self.outputs_record_schema = cast(
+                CWLObjectType,
+                make_valid_avro(self.outputs_record_schema, {}, set()),
+            )
+            make_avsc_object(convert_to_dict(self.outputs_record_schema), self.names)
+
+        if toolpath_object.get("class") is not None and not getdefault(
+            loadingContext.disable_js_validation, False
+        ):
+            validate_js_options = (
+                None
+            )  # type: Optional[Dict[str, Union[List[str], str, int]]]
+            if loadingContext.js_hint_options_file is not None:
+                try:
+                    with open(loadingContext.js_hint_options_file) as options_file:
+                        validate_js_options = json.load(options_file)
+                except (OSError, ValueError):
+                    _logger.error(
+                        "Failed to read options file %s",
+                        loadingContext.js_hint_options_file,
+                    )
+                    raise
+            if self.doc_schema is not None:
+                validate_js_expressions(
+                    toolpath_object,
+                    self.doc_schema.names[toolpath_object["class"]],
+                    validate_js_options,
+                )
+
+        dockerReq, is_req = self.get_requirement("DockerRequirement")
+
+        if (
+            dockerReq is not None
+            and "dockerOutputDirectory" in dockerReq
+            and is_req is not None
+            and not is_req
+        ):
+            _logger.warning(
+                SourceLine(item=dockerReq, raise_type=str).makeError(
+                    "When 'dockerOutputDirectory' is declared, DockerRequirement "
+                    "should go in the 'requirements' section, not 'hints'."
+                    ""
+                )
+            )
+
+        if (
+            dockerReq is not None
+            and is_req is not None
+            and dockerReq.get("dockerOutputDirectory") == "/var/spool/cwl"
+        ):
+            if is_req:
+                # In this specific case, it is legal to have /var/spool/cwl, so skip the check.
+                pass
+            else:
+                # Must be a requirement
+                var_spool_cwl_detector(self.tool)
+        else:
+            var_spool_cwl_detector(self.tool)
+
+    def _init_job(
+        self, joborder: CWLObjectType, runtime_context: RuntimeContext
+    ) -> Builder:
+
+        if self.metadata.get("cwlVersion") != INTERNAL_VERSION:
+            raise WorkflowException(
+                "Process object loaded with version '%s', must update to '%s' in order to execute."
+                % (self.metadata.get("cwlVersion"), INTERNAL_VERSION)
+            )
+
+        job = copy.deepcopy(joborder)
+
+        make_fs_access = getdefault(runtime_context.make_fs_access, StdFsAccess)
+        fs_access = make_fs_access(runtime_context.basedir)
+
+        load_listing_req, _ = self.get_requirement("LoadListingRequirement")
+
+        load_listing = (
+            cast(str, load_listing_req.get("loadListing"))
+            if load_listing_req is not None
+            else "no_listing"
+        )
+
+        # Validate job order
+        try:
+            fill_in_defaults(self.tool["inputs"], job, fs_access)
+
+            normalizeFilesDirs(job)
+            schema = self.names.get_name("input_record_schema", None)
+            if schema is None:
+                raise WorkflowException(
+                    "Missing input record schema: " "{}".format(self.names)
+                )
+            validate_ex(schema, job, strict=False, logger=_logger_validation_warnings)
+
+            if load_listing and load_listing != "no_listing":
+                get_listing(fs_access, job, recursive=(load_listing == "deep_listing"))
+
+            visit_class(job, ("File",), functools.partial(add_sizes, fs_access))
+
+            if load_listing == "deep_listing":
+                for i, inparm in enumerate(self.tool["inputs"]):
+                    k = shortname(inparm["id"])
+                    if k not in job:
+                        continue
+                    v = job[k]
+                    dircount = [0]
+
+                    def inc(d):  # type: (List[int]) -> None
+                        d[0] += 1
+
+                    visit_class(v, ("Directory",), lambda x: inc(dircount))
+                    if dircount[0] == 0:
+                        continue
+                    filecount = [0]
+                    visit_class(v, ("File",), lambda x: inc(filecount))
+                    if filecount[0] > FILE_COUNT_WARNING:
+                        # Long lines in this message are okay, will be reflowed based on terminal columns.
+                        _logger.warning(
+                            strip_dup_lineno(
+                                SourceLine(self.tool["inputs"], i, str).makeError(
+                                    """Recursive directory listing has resulted in a large number of File objects (%s) passed to the input parameter '%s'.  This may negatively affect workflow performance and memory use.
+
+If this is a problem, use the hint 'cwltool:LoadListingRequirement' with "shallow_listing" or "no_listing" to change the directory listing behavior:
+
+$namespaces:
+  cwltool: "http://commonwl.org/cwltool#"
+hints:
+  cwltool:LoadListingRequirement:
+    loadListing: shallow_listing
+
+"""
+                                    % (filecount[0], k)
+                                )
+                            )
+                        )
+
+        except (ValidationException, WorkflowException) as err:
+            raise WorkflowException("Invalid job input record:\n" + str(err)) from err
+
+        files = []  # type: List[CWLObjectType]
+        bindings = CommentedSeq()
+        outdir = ""
+        tmpdir = ""
+        stagedir = ""
+
+        docker_req, _ = self.get_requirement("DockerRequirement")
+        default_docker = None
+
+        if docker_req is None and runtime_context.default_container:
+            default_docker = runtime_context.default_container
+
+        if (docker_req or default_docker) and runtime_context.use_container:
+            if docker_req is not None:
+                # Check if docker output directory is absolute
+                if docker_req.get("dockerOutputDirectory") and cast(
+                    str, docker_req.get("dockerOutputDirectory")
+                ).startswith("/"):
+                    outdir = cast(str, docker_req.get("dockerOutputDirectory"))
+                else:
+                    outdir = cast(
+                        str,
+                        docker_req.get("dockerOutputDirectory")
+                        or runtime_context.docker_outdir
+                        or random_outdir(),
+                    )
+            elif default_docker is not None:
+                outdir = runtime_context.docker_outdir or random_outdir()
+            tmpdir = runtime_context.docker_tmpdir or "/tmp"  # nosec
+            stagedir = runtime_context.docker_stagedir or "/var/lib/cwl"
+        else:
+            if self.tool["class"] == "CommandLineTool":
+                outdir = fs_access.realpath(runtime_context.get_outdir())
+                tmpdir = fs_access.realpath(runtime_context.get_tmpdir())
+                stagedir = fs_access.realpath(runtime_context.get_stagedir())
+
+        cwl_version = cast(
+            str,
+            self.metadata.get("http://commonwl.org/cwltool#original_cwlVersion", None),
+        )
+        builder = Builder(
+            job,
+            files,
+            bindings,
+            self.schemaDefs,
+            self.names,
+            self.requirements,
+            self.hints,
+            {},
+            runtime_context.mutation_manager,
+            self.formatgraph,
+            make_fs_access,
+            fs_access,
+            runtime_context.job_script_provider,
+            runtime_context.eval_timeout,
+            runtime_context.debug,
+            runtime_context.js_console,
+            runtime_context.force_docker_pull,
+            load_listing,
+            outdir,
+            tmpdir,
+            stagedir,
+            cwl_version,
+        )
+
+        bindings.extend(
+            builder.bind_input(
+                self.inputs_record_schema,
+                job,
+                discover_secondaryFiles=getdefault(runtime_context.toplevel, False),
+            )
+        )
+
+        if self.tool.get("baseCommand"):
+            for index, command in enumerate(aslist(self.tool["baseCommand"])):
+                bindings.append({"position": [-1000000, index], "datum": command})
+
+        if self.tool.get("arguments"):
+            for i, arg in enumerate(self.tool["arguments"]):
+                lc = self.tool["arguments"].lc.data[i]
+                filename = self.tool["arguments"].lc.filename
+                bindings.lc.add_kv_line_col(len(bindings), lc)
+                if isinstance(arg, MutableMapping):
+                    arg = copy.deepcopy(arg)
+                    if arg.get("position"):
+                        position = arg.get("position")
+                        if isinstance(position, str):  # no need to test the
+                            # CWLVersion as the v1.0
+                            # schema only allows ints
+                            position = builder.do_eval(position)
+                            if position is None:
+                                position = 0
+                        arg["position"] = [position, i]
+                    else:
+                        arg["position"] = [0, i]
+                    bindings.append(arg)
+                elif ("$(" in arg) or ("${" in arg):
+                    cm = CommentedMap((("position", [0, i]), ("valueFrom", arg)))
+                    cm.lc.add_kv_line_col("valueFrom", lc)
+                    cm.lc.filename = filename
+                    bindings.append(cm)
+                else:
+                    cm = CommentedMap((("position", [0, i]), ("datum", arg)))
+                    cm.lc.add_kv_line_col("datum", lc)
+                    cm.lc.filename = filename
+                    bindings.append(cm)
+
+        # use python2 like sorting of heterogeneous lists
+        # (containing str and int types),
+        key = functools.cmp_to_key(cmp_like_py2)
+
+        # This awkward construction replaces the contents of
+        # "bindings" in place (because Builder expects it to be
+        # mutated in place, sigh, I'm sorry) with its contents sorted,
+        # supporting different versions of Python and ruamel.yaml with
+        # different behaviors/bugs in CommentedSeq.
+        bindings_copy = copy.deepcopy(bindings)
+        del bindings[:]
+        bindings.extend(sorted(bindings_copy, key=key))
+
+        if self.tool["class"] != "Workflow":
+            builder.resources = self.evalResources(builder, runtime_context)
+        return builder
+
+    def evalResources(
+        self, builder: Builder, runtimeContext: RuntimeContext
+    ) -> Dict[str, Union[int, float, str]]:
+        resourceReq, _ = self.get_requirement("ResourceRequirement")
+        if resourceReq is None:
+            resourceReq = {}
+        cwl_version = self.metadata.get(
+            "http://commonwl.org/cwltool#original_cwlVersion", None
+        )
+        if cwl_version == "v1.0":
+            ram = 1024
+        else:
+            ram = 256
+        request: Dict[str, Union[int, float, str]] = {
+            "coresMin": 1,
+            "coresMax": 1,
+            "ramMin": ram,
+            "ramMax": ram,
+            "tmpdirMin": 1024,
+            "tmpdirMax": 1024,
+            "outdirMin": 1024,
+            "outdirMax": 1024,
+        }
+        for a in ("cores", "ram", "tmpdir", "outdir"):
+            mn = mx = None  # type: Optional[Union[int, float]]
+            if resourceReq.get(a + "Min"):
+                mn = cast(
+                    Union[int, float],
+                    eval_resource(
+                        builder, cast(Union[str, int, float], resourceReq[a + "Min"])
+                    ),
+                )
+            if resourceReq.get(a + "Max"):
+                mx = cast(
+                    Union[int, float],
+                    eval_resource(
+                        builder, cast(Union[str, int, float], resourceReq[a + "Max"])
+                    ),
+                )
+            if mn is None:
+                mn = mx
+            elif mx is None:
+                mx = mn
+
+            if mn is not None:
+                request[a + "Min"] = mn
+                request[a + "Max"] = cast(Union[int, float], mx)
+
+        if runtimeContext.select_resources is not None:
+            return runtimeContext.select_resources(request, runtimeContext)
+        return {
+            "cores": request["coresMin"],
+            "ram": math.ceil(request["ramMin"])
+            if not isinstance(request["ramMin"], str)
+            else request["ramMin"],
+            "tmpdirSize": math.ceil(request["tmpdirMin"])
+            if not isinstance(request["tmpdirMin"], str)
+            else request["tmpdirMin"],
+            "outdirSize": math.ceil(request["outdirMin"])
+            if not isinstance(request["outdirMin"], str)
+            else request["outdirMin"],
+        }
+
+    def validate_hints(
+        self, avsc_names: Names, hints: List[CWLObjectType], strict: bool
+    ) -> None:
+        for i, r in enumerate(hints):
+            sl = SourceLine(hints, i, ValidationException)
+            with sl:
+                if (
+                    avsc_names.get_name(cast(str, r["class"]), None) is not None
+                    and self.doc_loader is not None
+                ):
+                    plain_hint = {
+                        key: r[key]
+                        for key in r
+                        if key not in self.doc_loader.identifiers
+                    }  # strip identifiers
+                    validate_ex(
+                        cast(
+                            Schema,
+                            avsc_names.get_name(cast(str, plain_hint["class"]), None),
+                        ),
+                        plain_hint,
+                        strict=strict,
+                    )
+                elif r["class"] in ("NetworkAccess", "LoadListingRequirement"):
+                    pass
+                else:
+                    _logger.info(str(sl.makeError("Unknown hint %s" % (r["class"]))))
+
+    def visit(self, op: Callable[[CommentedMap], None]) -> None:
+        op(self.tool)
+
+    @abc.abstractmethod
+    def job(
+        self,
+        job_order: CWLObjectType,
+        output_callbacks: Optional[OutputCallbackType],
+        runtimeContext: RuntimeContext,
+    ) -> JobsGeneratorType:
+        pass
+
+
+_names = set()  # type: Set[str]
+
+
+def uniquename(stem: str, names: Optional[Set[str]] = None) -> str:
+    global _names
+    if names is None:
+        names = _names
+    c = 1
+    u = stem
+    while u in names:
+        c += 1
+        u = f"{stem}_{c}"
+    names.add(u)
+    return u
+
+
+def nestdir(base: str, deps: CWLObjectType) -> CWLObjectType:
+    dirname = os.path.dirname(base) + "/"
+    subid = cast(str, deps["location"])
+    if subid.startswith(dirname):
+        s2 = subid[len(dirname) :]
+        sp = s2.split("/")
+        sp.pop()
+        while sp:
+            nx = sp.pop()
+            deps = {"class": "Directory", "basename": nx, "listing": [deps]}
+    return deps
+
+
+def mergedirs(listing: List[CWLObjectType]) -> List[CWLObjectType]:
+    r = []  # type: List[CWLObjectType]
+    ents = {}  # type: Dict[str, CWLObjectType]
+    collided = set()  # type: Set[str]
+    for e in listing:
+        basename = cast(str, e["basename"])
+        if basename not in ents:
+            ents[basename] = e
+        elif e["class"] == "Directory":
+            if e.get("listing"):
+                cast(
+                    List[CWLObjectType], ents[basename].setdefault("listing", [])
+                ).extend(cast(List[CWLObjectType], e["listing"]))
+            if cast(str, ents[basename]["location"]).startswith("_:"):
+                ents[basename]["location"] = e["location"]
+        elif e["location"] != ents[basename]["location"]:
+            # same basename, different location, collision,
+            # rename both.
+            collided.add(basename)
+            e2 = ents[basename]
+
+            e["basename"] = urllib.parse.quote(cast(str, e["location"]), safe="")
+            e2["basename"] = urllib.parse.quote(cast(str, e2["location"]), safe="")
+
+            e["nameroot"], e["nameext"] = os.path.splitext(cast(str, e["basename"]))
+            e2["nameroot"], e2["nameext"] = os.path.splitext(cast(str, e2["basename"]))
+
+            ents[cast(str, e["basename"])] = e
+            ents[cast(str, e2["basename"])] = e2
+    for c in collided:
+        del ents[c]
+    for e in ents.values():
+        if e["class"] == "Directory" and "listing" in e:
+            e["listing"] = cast(
+                MutableSequence[CWLOutputAtomType],
+                mergedirs(cast(List[CWLObjectType], e["listing"])),
+            )
+    r.extend(ents.values())
+    return r
+
+
+CWL_IANA = "https://www.iana.org/assignments/media-types/application/cwl"
+
+
+def scandeps(
+    base: str,
+    doc: Union[CWLObjectType, MutableSequence[CWLObjectType]],
+    reffields: Set[str],
+    urlfields: Set[str],
+    loadref: Callable[[str, str], Union[CommentedMap, CommentedSeq, str, None]],
+    urljoin: Callable[[str, str], str] = urllib.parse.urljoin,
+    nestdirs: bool = True,
+) -> MutableSequence[CWLObjectType]:
+    r = []  # type: MutableSequence[CWLObjectType]
+    if isinstance(doc, MutableMapping):
+        if "id" in doc:
+            if cast(str, doc["id"]).startswith("file://"):
+                df, _ = urllib.parse.urldefrag(cast(str, doc["id"]))
+                if base != df:
+                    r.append({"class": "File", "location": df, "format": CWL_IANA})
+                    base = df
+
+        if doc.get("class") in ("File", "Directory") and "location" in urlfields:
+            u = cast(Optional[str], doc.get("location", doc.get("path")))
+            if u and not u.startswith("_:"):
+                deps = {
+                    "class": doc["class"],
+                    "location": urljoin(base, u),
+                }  # type: CWLObjectType
+                if "basename" in doc:
+                    deps["basename"] = doc["basename"]
+                if doc["class"] == "Directory" and "listing" in doc:
+                    deps["listing"] = doc["listing"]
+                if doc["class"] == "File" and "secondaryFiles" in doc:
+                    deps["secondaryFiles"] = cast(
+                        CWLOutputAtomType,
+                        scandeps(
+                            base,
+                            cast(
+                                Union[CWLObjectType, MutableSequence[CWLObjectType]],
+                                doc["secondaryFiles"],
+                            ),
+                            reffields,
+                            urlfields,
+                            loadref,
+                            urljoin=urljoin,
+                            nestdirs=nestdirs,
+                        ),
+                    )
+                if nestdirs:
+                    deps = nestdir(base, deps)
+                r.append(deps)
+            else:
+                if doc["class"] == "Directory" and "listing" in doc:
+                    r.extend(
+                        scandeps(
+                            base,
+                            cast(MutableSequence[CWLObjectType], doc["listing"]),
+                            reffields,
+                            urlfields,
+                            loadref,
+                            urljoin=urljoin,
+                            nestdirs=nestdirs,
+                        )
+                    )
+                elif doc["class"] == "File" and "secondaryFiles" in doc:
+                    r.extend(
+                        scandeps(
+                            base,
+                            cast(MutableSequence[CWLObjectType], doc["secondaryFiles"]),
+                            reffields,
+                            urlfields,
+                            loadref,
+                            urljoin=urljoin,
+                            nestdirs=nestdirs,
+                        )
+                    )
+
+        for k, v in doc.items():
+            if k in reffields:
+                for u2 in aslist(v):
+                    if isinstance(u2, MutableMapping):
+                        r.extend(
+                            scandeps(
+                                base,
+                                u2,
+                                reffields,
+                                urlfields,
+                                loadref,
+                                urljoin=urljoin,
+                                nestdirs=nestdirs,
+                            )
+                        )
+                    else:
+                        subid = urljoin(base, u2)
+                        basedf, _ = urllib.parse.urldefrag(base)
+                        subiddf, _ = urllib.parse.urldefrag(subid)
+                        if basedf == subiddf:
+                            continue
+                        sub = cast(
+                            Union[MutableSequence[CWLObjectType], CWLObjectType],
+                            loadref(base, u2),
+                        )
+                        deps2 = {
+                            "class": "File",
+                            "location": subid,
+                            "format": CWL_IANA,
+                        }  # type: CWLObjectType
+                        sf = scandeps(
+                            subid,
+                            sub,
+                            reffields,
+                            urlfields,
+                            loadref,
+                            urljoin=urljoin,
+                            nestdirs=nestdirs,
+                        )
+                        if sf:
+                            deps2["secondaryFiles"] = cast(
+                                MutableSequence[CWLOutputAtomType], sf
+                            )
+                        if nestdirs:
+                            deps2 = nestdir(base, deps2)
+                        r.append(deps2)
+            elif k in urlfields and k != "location":
+                for u3 in aslist(v):
+                    deps = {"class": "File", "location": urljoin(base, u3)}
+                    if nestdirs:
+                        deps = nestdir(base, deps)
+                    r.append(deps)
+            elif doc.get("class") in ("File", "Directory") and k in (
+                "listing",
+                "secondaryFiles",
+            ):
+                # should be handled earlier.
+                pass
+            else:
+                r.extend(
+                    scandeps(
+                        base,
+                        cast(Union[MutableSequence[CWLObjectType], CWLObjectType], v),
+                        reffields,
+                        urlfields,
+                        loadref,
+                        urljoin=urljoin,
+                        nestdirs=nestdirs,
+                    )
+                )
+    elif isinstance(doc, MutableSequence):
+        for d in doc:
+            r.extend(
+                scandeps(
+                    base,
+                    d,
+                    reffields,
+                    urlfields,
+                    loadref,
+                    urljoin=urljoin,
+                    nestdirs=nestdirs,
+                )
+            )
+
+    if r:
+        normalizeFilesDirs(r)
+        r = mergedirs(cast(List[CWLObjectType], r))
+
+    return r
+
+
+def compute_checksums(fs_access: StdFsAccess, fileobj: CWLObjectType) -> None:
+    if "checksum" not in fileobj:
+        checksum = hashlib.sha1()  # nosec
+        location = cast(str, fileobj["location"])
+        with fs_access.open(location, "rb") as f:
+            contents = f.read(1024 * 1024)
+            while contents != b"":
+                checksum.update(contents)
+                contents = f.read(1024 * 1024)
+        fileobj["checksum"] = "sha1$%s" % checksum.hexdigest()
+        fileobj["size"] = fs_access.size(location)