Add recursive subworkflow linting and output source validation

Recurse into inline subworkflows (run: {class: GalaxyWorkflow})
in lint_format2, matching lint_ga's existing recursion pattern.

Add _validate_output_sources to catch dangling outputSource refs
pointing to nonexistent steps — fixes the nested_no_steps bug.

Add LintContext.child(prefix) for nested error context paths
e.g. "[step nested_workflow] Output 'x' references step 'y'..."

Closes the xfail test — linter now catches exit code 2.
See #162 for expanded lint mode (URL/@import subworkflows).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: jmchilton

gxformat2/lint.py

@@ -118,16 +118,58 @@ def lint_format2(lint_context, workflow_dict, path=None):
         lint_context.error("Validation failed " + str(e))
 
     steps = ensure_key_if_present(lint_context, workflow_dict, "steps", default={}, has_class=(dict, list))
-    steps = steps.values() if isinstance(steps, dict) else steps
-    for step in steps:
+    steps_items = list(walk_id_list_or_dict(steps)) if steps else []
+    for step_key, step in steps_items:
         _lint_step_errors(lint_context, step)
         _lint_tool_if_present(lint_context, step)
 
+        # Recurse into inline subworkflows
+        run = step.get("run")
+        if isinstance(run, dict) and run.get("class") == "GalaxyWorkflow":
+            step_label = step.get("label") or step_key or "?"
+            child_ctx = lint_context.child(f"step {step_label}")
+            lint_format2(child_ctx, run, path=path)
+
+    _validate_output_sources(lint_context, workflow_dict, steps_items)
     _validate_input_types(lint_context, workflow_dict)
     _validate_report(lint_context, workflow_dict)
     _lint_training(lint_context, workflow_dict)
 
 
+def _validate_output_sources(lint_context, workflow_dict, steps_items):
+    """Check that outputSource references point to existing step/input labels."""
+    step_labels = set()
+    for step_key, step in steps_items:
+        label = step.get("label") or step_key
+        if label is not None:
+            step_labels.add(str(label))
+
+    # Also include input labels
+    inputs = workflow_dict.get("inputs")
+    if inputs:
+        for input_key, inp in walk_id_list_or_dict(inputs):
+            label = inp.get("label") or input_key if isinstance(inp, dict) else input_key
+            if label is not None:
+                step_labels.add(str(label))
+
+    outputs = workflow_dict.get("outputs")
+    if not outputs:
+        return
+    for output_key, output in walk_id_list_or_dict(outputs):
+        if not isinstance(output, dict):
+            continue
+        output_source = output.get("outputSource")
+        if not output_source or not isinstance(output_source, str):
+            continue
+        # Extract step reference — format is "step_label/output_name" or just "step_label"
+        step_ref = output_source.split("/")[0]
+        if step_ref not in step_labels:
+            lint_context.error(
+                f"Output '{output_key}' references step '{step_ref}' via outputSource "
+                f"'{output_source}', but no step or input with that label exists"
+            )
+
+
 def _validate_input_types(lint_context: LintContext, workflow_dict: dict):
     try:
         inputs = Inputs(workflow_dict)

gxformat2/linting.py

@@ -12,22 +12,31 @@
 class LintContext:
     """Track running status (state) of linting."""
 
-    def __init__(self, level=LEVEL_WARN, training_topic=DEFAULT_TRAINING_LINT):
+    def __init__(self, level=LEVEL_WARN, training_topic=DEFAULT_TRAINING_LINT, _prefix=""):
         """Create LintContext with specified 'level' (currently unused)."""
         self.level = level
         self.training_topic = training_topic
         self.found_errors = False
         self.found_warns = False
+        self._prefix = _prefix
 
         # self.valid_messages = []
         # self.info_messages = []
         self.warn_messages = []
         self.error_messages = []
 
+    def child(self, prefix):
+        """Create child context that prefixes messages and shares parent message lists."""
+        full_prefix = f"{self._prefix}[{prefix}] " if not self._prefix else f"{self._prefix[:-1]} > {prefix}] "
+        child_ctx = LintContext(level=self.level, training_topic=self.training_topic, _prefix=full_prefix)
+        child_ctx.warn_messages = self.warn_messages
+        child_ctx.error_messages = self.error_messages
+        return child_ctx
+
     def __handle_message(self, message_list, message, *args, **kwds):
         if kwds or args:
             message = message.format(*args, **kwds)
-        message_list.append(message)
+        message_list.append(f"{self._prefix}{message}")
 
     # def valid(self, message, *args, **kwds):
     #     self.__handle_message(self.valid_messages, message, *args, **kwds)

tests/test_lint.py

@@ -172,14 +172,7 @@ def setup_module(module):
 
     invalid_format2_nested = _deep_copy(green_format2_nested)
     del invalid_format2_nested["steps"]["nested_workflow"]["run"]["steps"]
-    # BUG: should be exit code 2 but widening WorkflowStep.run to Any?
-    # (for URL/import support in d7e5dae) means neither schema-salad nor
-    # pydantic validates nested subworkflow structure. The subworkflow dict
-    # passes as an opaque dict instead of being validated as GalaxyWorkflow.
-    # Normalization also doesn't catch it — the dict stays raw and to_native
-    # crashes with a label lookup failure. Needs a lint rule that validates
-    # inline subworkflow dicts independently.
-    _dump_with_exit_code(invalid_format2_nested, 0, "format2_nested_no_steps")
+    _dump_with_exit_code(invalid_format2_nested, 2, "format2_nested_no_steps")
 
     invalid_native_nested = _deep_copy(green_native_nested)
     del invalid_native_nested["steps"]["2"]["subworkflow"]["steps"]
@@ -265,8 +258,6 @@ def setup_module(module):
     # ensure that round tripping all green format2 workflows still lint green.
     for file_name in os.listdir(TEST_LINT_EXAMPLES):
         if file_name.startswith("0_format2") and "roundtrip" not in file_name:
-            if "no_steps" in file_name:
-                continue  # tested separately via test_roundtrip_nested_no_steps
             roundtrip_contents = round_trip(open(os.path.join(TEST_LINT_EXAMPLES, file_name)).read())
             base = os.path.splitext(file_name)[0][len("0_") :]
             _dump_with_exit_code(roundtrip_contents, 0, base + "_roundtrip")
@@ -275,19 +266,6 @@ def setup_module(module):
 SKIP_BP = "--skip-best-practices"
 
 
-@pytest.mark.xfail(
-    reason="WorkflowStep.run widened to Any? for URL/import support — "
-    "inline subworkflow dicts are not validated as GalaxyWorkflow, "
-    "so missing steps is undetected. Needs a dedicated lint rule.",
-    raises=ValueError,
-)
-def test_roundtrip_nested_no_steps():
-    """Round-tripping a nested workflow with deleted steps should fail lint."""
-    path = os.path.join(TEST_LINT_EXAMPLES, "0_format2_nested_no_steps.yml")
-    if os.path.exists(path):
-        round_trip(open(path).read())
-
-
 def test_lint_ga_basic():
     assert main(["lint", SKIP_BP, example_path("real-shed-tools-raw.ga")]) == 1  # no outputs