common-workflow-language · mr-c · Dec 18, 2025 · Oct 30, 2025 · Dec 8, 2025 · Dec 16, 2025
diff --git a/cwl_utils/cwl_v1_0_expression_refactor.py b/cwl_utils/cwl_v1_0_expression_refactor.py
@@ -7,7 +7,7 @@
 import uuid
 from collections.abc import MutableSequence, Sequence
 from contextlib import suppress
-from typing import Any, Optional, cast
+from typing import Any, cast
 
 from ruamel import yaml
 from schema_salad.sourceline import SourceLine
@@ -445,7 +445,7 @@ def find_expressionLib(
         if process.requirements:
             for req in process.requirements:
                 if isinstance(req, cwl.InlineJavascriptRequirement):
-                    return cast(Optional[list[str]], copy.deepcopy(req.expressionLib))
+                    return cast(list[str] | None, copy.deepcopy(req.expressionLib))
     return None
 
 
@@ -1695,7 +1695,9 @@ def cltool_step_outputs_to_workflow_outputs(
 def generate_etool_from_expr2(
     expr: str,
     target: cwl.InputParameter,
-    inputs: Sequence[cwl.InputParameter | cwl.CommandInputParameter],
+    inputs: Sequence[
+        cwl.InputParameter | cwl.CommandInputParameter | cwl.CommandOutputParameter
+    ],
     self_name: str | None = None,
     process: cwl.CommandLineTool | cwl.ExpressionTool | None = None,
     extra_processes: None | (
@@ -1815,9 +1817,12 @@ def traverse_step(
                 if not target:
                     raise WorkflowException("target not found")
                 input_source_id = None
-                source_type: None | (list[cwl.InputParameter] | cwl.InputParameter) = (
+                source_type: (
                     None
-                )
+                    | MutableSequence[cwl.InputParameter | cwl.CommandOutputParameter]
+                    | cwl.InputParameter
+                    | cwl.CommandOutputParameter
+                ) = None
                 if inp.source:
                     if isinstance(inp.source, MutableSequence):
                         input_source_id = []
@@ -1904,7 +1909,7 @@ def traverse_step(
 
 def workflow_step_to_InputParameters(
     step_ins: list[cwl.WorkflowStepInput], parent: cwl.Workflow, except_in_id: str
-) -> list[cwl.InputParameter]:
+) -> list[cwl.InputParameter | cwl.CommandOutputParameter]:
     """Create InputParameters to match the given WorkflowStep inputs."""
     params = []
     for inp in step_ins:
@@ -1915,7 +1920,7 @@ def workflow_step_to_InputParameters(
             param = copy.deepcopy(
                 utils.param_for_source_id(parent, sourcenames=inp.source)
             )
-            if isinstance(param, list):
+            if isinstance(param, MutableSequence):
                 for p in param:
                     if not p.type_:
                         raise WorkflowException(
@@ -1946,7 +1951,12 @@ def replace_step_valueFrom_expr_with_etool(
     original_step_ins: list[cwl.WorkflowStepInput],
     source: str | list[str] | None,
     replace_etool: bool,
-    source_type: cwl.InputParameter | list[cwl.InputParameter] | None = None,
+    source_type: (
+        cwl.InputParameter
+        | cwl.CommandOutputParameter
+        | MutableSequence[cwl.InputParameter | cwl.CommandOutputParameter]
+        | None
+    ) = None,
 ) -> None:
     """Replace a WorkflowStep level 'valueFrom' expression with a sibling ExpressionTool step."""
     if not step_inp.id:

diff --git a/cwl_utils/cwl_v1_1_expression_refactor.py b/cwl_utils/cwl_v1_1_expression_refactor.py
@@ -7,7 +7,7 @@
 import uuid
 from collections.abc import MutableSequence, Sequence
 from contextlib import suppress
-from typing import Any, Optional, cast
+from typing import Any, cast
 
 from ruamel import yaml
 from schema_salad.sourceline import SourceLine
@@ -443,7 +443,7 @@ def find_expressionLib(
         if process.requirements:
             for req in process.requirements:
                 if isinstance(req, cwl.InlineJavascriptRequirement):
-                    return cast(Optional[list[str]], copy.deepcopy(req.expressionLib))
+                    return cast(list[str] | None, copy.deepcopy(req.expressionLib))
     return None
 
 
@@ -1697,7 +1697,11 @@ def cltool_step_outputs_to_workflow_outputs(
 def generate_etool_from_expr2(
     expr: str,
     target: cwl.CommandInputParameter | cwl.WorkflowInputParameter,
-    inputs: Sequence[cwl.WorkflowInputParameter | cwl.CommandInputParameter],
+    inputs: Sequence[
+        cwl.WorkflowInputParameter
+        | cwl.CommandInputParameter
+        | cwl.CommandOutputParameter
+    ],
     self_name: str | None = None,
     process: cwl.CommandLineTool | cwl.ExpressionTool | None = None,
     extra_processes: None | (
@@ -1818,7 +1822,14 @@ def traverse_step(
                     raise WorkflowException("target not found")
                 input_source_id = None
                 source_type: None | (
-                    list[cwl.WorkflowInputParameter] | cwl.WorkflowInputParameter
+                    MutableSequence[
+                        cwl.CommandInputParameter
+                        | cwl.CommandOutputParameter
+                        | cwl.WorkflowInputParameter
+                    ]
+                    | cwl.CommandInputParameter
+                    | cwl.CommandOutputParameter
+                    | cwl.WorkflowInputParameter
                 ) = None
                 if inp.source:
                     if isinstance(inp.source, MutableSequence):
@@ -1906,7 +1917,9 @@ def traverse_step(
 
 def workflow_step_to_WorkflowInputParameters(
     step_ins: list[cwl.WorkflowStepInput], parent: cwl.Workflow, except_in_id: str
-) -> list[cwl.WorkflowInputParameter]:
+) -> MutableSequence[
+    cwl.CommandInputParameter | cwl.CommandOutputParameter | cwl.WorkflowInputParameter
+]:
     """Create WorkflowInputParameters to match the given WorkflowStep inputs."""
     params = []
     for inp in step_ins:
@@ -1917,7 +1930,7 @@ def workflow_step_to_WorkflowInputParameters(
             param = copy.deepcopy(
                 utils.param_for_source_id(parent, sourcenames=inp.source)
             )
-            if isinstance(param, list):
+            if isinstance(param, MutableSequence):
                 for p in param:
                     p.id = inp_id
                     p.type_ = clean_type_ids(p.type_)
@@ -1941,7 +1954,14 @@ def replace_step_valueFrom_expr_with_etool(
     source: str | list[str] | None,
     replace_etool: bool,
     source_type: None | (
-        cwl.WorkflowInputParameter | list[cwl.WorkflowInputParameter]
+        cwl.CommandInputParameter
+        | cwl.CommandOutputParameter
+        | cwl.WorkflowInputParameter
+        | MutableSequence[
+            cwl.CommandInputParameter
+            | cwl.CommandOutputParameter
+            | cwl.WorkflowInputParameter
+        ]
     ) = None,
 ) -> None:
     """Replace a WorkflowStep level 'valueFrom' expression with a sibling ExpressionTool step."""

diff --git a/cwl_utils/cwl_v1_2_expression_refactor.py b/cwl_utils/cwl_v1_2_expression_refactor.py
@@ -7,7 +7,7 @@
 import uuid
 from collections.abc import Mapping, MutableSequence, Sequence
 from contextlib import suppress
-from typing import Any, Optional, cast
+from typing import Any, cast
 
 from ruamel import yaml
 from schema_salad.sourceline import SourceLine
@@ -442,7 +442,7 @@ def find_expressionLib(
         if process.requirements:
             for req in process.requirements:
                 if isinstance(req, cwl.InlineJavascriptRequirement):
-                    return cast(Optional[list[str]], copy.deepcopy(req.expressionLib))
+                    return cast(list[str] | None, copy.deepcopy(req.expressionLib))
     return None
 
 
@@ -725,7 +725,9 @@ def process_workflow_inputs_and_outputs(
                         source_type_items.append("null")
                 elif source_type_items != "null":
                     source_type_items = ["null", source_type_items]
-                source_type = cwl.CommandInputParameter(type_=source_type_items)
+                source_type = cwl.CommandInputParameter(
+                    id=None, type_=source_type_items
+                )
                 replace_expr_with_etool(
                     expression,
                     etool_id,
@@ -1798,7 +1800,11 @@ def cltool_step_outputs_to_workflow_outputs(
 def generate_etool_from_expr2(
     expr: str,
     target: cwl.CommandInputParameter | cwl.WorkflowInputParameter,
-    inputs: Sequence[cwl.WorkflowInputParameter | cwl.CommandInputParameter],
+    inputs: Sequence[
+        cwl.WorkflowInputParameter
+        | cwl.CommandInputParameter
+        | cwl.CommandOutputParameter
+    ],
     self_name: str | None = None,
     process: cwl.CommandLineTool | cwl.ExpressionTool | None = None,
     extra_processes: None | (
@@ -1919,7 +1925,14 @@ def traverse_step(
                     raise WorkflowException("target not found")
                 input_source_id = None
                 source_type: None | (
-                    list[cwl.WorkflowInputParameter] | cwl.WorkflowInputParameter
+                    MutableSequence[
+                        cwl.CommandInputParameter
+                        | cwl.CommandOutputParameter
+                        | cwl.WorkflowInputParameter
+                    ]
+                    | cwl.CommandInputParameter
+                    | cwl.CommandOutputParameter
+                    | cwl.WorkflowInputParameter
                 ) = None
                 if inp.source:
                     if isinstance(inp.source, MutableSequence):
@@ -2015,7 +2028,9 @@ def traverse_step(
 
 def workflow_step_to_WorkflowInputParameters(
     step_ins: list[cwl.WorkflowStepInput], parent: cwl.Workflow, except_in_id: str
-) -> list[cwl.WorkflowInputParameter]:
+) -> MutableSequence[
+    cwl.CommandInputParameter | cwl.CommandOutputParameter | cwl.WorkflowInputParameter
+]:
     """Create WorkflowInputParameters to match the given WorkflowStep inputs."""
     params = []
     for inp in step_ins:
@@ -2026,7 +2041,7 @@ def workflow_step_to_WorkflowInputParameters(
             param = copy.deepcopy(
                 utils.param_for_source_id(parent, sourcenames=inp.source)
             )
-            if isinstance(param, list):
+            if isinstance(param, MutableSequence):
                 for p in param:
                     p.id = inp_id
                     p.type_ = clean_type_ids(p.type_)
@@ -2050,7 +2065,14 @@ def replace_step_valueFrom_expr_with_etool(
     source: str | list[str] | None,
     replace_etool: bool,
     source_type: None | (
-        cwl.WorkflowInputParameter | list[cwl.WorkflowInputParameter]
+        cwl.CommandInputParameter
+        | cwl.CommandOutputParameter
+        | cwl.WorkflowInputParameter
+        | MutableSequence[
+            cwl.CommandInputParameter
+            | cwl.CommandOutputParameter
+            | cwl.WorkflowInputParameter
+        ]
     ) = None,
 ) -> None:
     """Replace a WorkflowStep level 'valueFrom' expression with a sibling ExpressionTool step."""

diff --git a/cwl_utils/expression.py b/cwl_utils/expression.py
@@ -1,12 +1,11 @@
 # SPDX-License-Identifier: Apache-2.0
 """CWL Expression parsing."""
 import asyncio
-import copy
 import inspect
 import json
-from collections.abc import Awaitable, MutableMapping
+from collections.abc import Awaitable, Container
 from enum import Enum
-from typing import Any, Union, cast
+from typing import Any, Final, cast
 
 from schema_salad.utils import json_dumps
 
@@ -21,6 +20,15 @@
 )
 from cwl_utils.utils import bytes2str_in_dicts
 
+OLD_ESCAPE_CWL_VERSIONS: Final[Container[str]] = (
+    "v1.0",
+    "v1.1.0-dev1",
+    "v1.1",
+    "v1.2.0-dev1",
+    "v1.2.0-dev2",
+    "v1.2.0-dev3",
+)
+
 
 def _convert_dumper(string: str) -> str:
     return f"{json.dumps(string)} + "
@@ -293,9 +301,7 @@ def do_eval(
 
     :param timeout: The maximum number of seconds to wait while executing.
     """
-    runtime = cast(MutableMapping[str, Union[int, str, None]], copy.deepcopy(resources))
-    runtime["tmpdir"] = tmpdir or None
-    runtime["outdir"] = outdir or None
+    runtime = resources | {"tmpdir": tmpdir or None, "outdir": outdir or None}
 
     rootvars = cast(
         CWLParameterContext,
@@ -319,19 +325,7 @@ def do_eval(
                 fullJS=fullJS,
                 jslib=jslib,
                 strip_whitespace=strip_whitespace,
-                escaping_behavior=(
-                    1
-                    if cwlVersion
-                    in (
-                        "v1.0",
-                        "v1.1.0-dev1",
-                        "v1.1",
-                        "v1.2.0-dev1",
-                        "v1.2.0-dev2",
-                        "v1.2.0-dev3",
-                    )
-                    else 2
-                ),
+                escaping_behavior=1 if cwlVersion in OLD_ESCAPE_CWL_VERSIONS else 2,
                 **kwargs,
             )
 

diff --git a/cwl_utils/expression_refactor.py b/cwl_utils/expression_refactor.py
@@ -8,7 +8,7 @@
 import sys
 from collections.abc import Callable, MutableMapping, MutableSequence
 from pathlib import Path
-from typing import Any, Optional, Protocol, Union
+from typing import Any, Protocol
 
 from ruamel.yaml.main import YAML
 from ruamel.yaml.scalarstring import walk_tree
@@ -28,9 +28,9 @@
 _logger.setLevel(logging.INFO)
 _cwlutilslogger.setLevel(100)
 
-save_type = Optional[
-    Union[MutableMapping[str, Any], MutableSequence[Any], int, float, bool, str]
-]
+save_type = (
+    MutableMapping[str, Any] | MutableSequence[Any] | int | float | bool | str | None
+)
 
 
 class saveCWL(Protocol):

diff --git a/cwl_utils/inputs_schema_gen.py b/cwl_utils/inputs_schema_gen.py
@@ -12,7 +12,7 @@
 from copy import deepcopy
 from importlib.resources import files
 from pathlib import Path
-from typing import Any, TypeGuard, Union
+from typing import Any, TypeGuard
 from urllib.parse import urlparse
 
 import requests
@@ -61,9 +61,9 @@
 }
 
 # Some type hinting
-InputType = Union[
-    InputArraySchema, InputEnumSchema, InputRecordSchema, str, File, Directory
-]
+InputType = (
+    InputArraySchema | InputEnumSchema | InputRecordSchema | str | File | Directory
+)
 
 
 # Don't need type checking at runtime