Remove dead connect syntax support, drop pop_connect_from_step_dict

The connect: key on format2 steps was never in the schema and has
zero usage in Galaxy, Planemo, or gxformat2 tests (only a 2016
slide deck). Remove handling from pre-clean and delete
pop_connect_from_step_dict from model.py. Rewrite tests to verify
$link resolution through normalization instead of old dict helpers.

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

Author: jmchilton

gxformat2/model.py

@@ -61,43 +61,6 @@ def get_native_step_type(gxformat2_step_dict: dict) -> _NativeGalaxyStepType:
     return step_type
 
 
-def pop_connect_from_step_dict(step: dict) -> dict:
-    """Merge 'in' and 'connect' keys into a unified connection dict separated from state.
-
-    Meant to be used an initial processing step in reasoning about connections defined by the
-    format2 step description.
-    """
-    if "connect" not in step:
-        step["connect"] = {}
-
-    connect = step["connect"]
-    del step["connect"]
-
-    # handle CWL-style in dict connections.
-    if "in" in step:
-        step_in = step["in"]
-        assert isinstance(step_in, dict)
-        connection_keys = set()
-        for key, value in step_in.items():
-            # TODO: this can be a list right?
-            if isinstance(value, dict) and "source" in value:
-                value = value["source"]
-            elif isinstance(value, dict) and "default" in value:
-                continue
-            elif isinstance(value, dict):
-                raise KeyError(f"step input must define either source or default {value}")
-            connect[key] = [value]
-            connection_keys.add(key)
-
-        for key in connection_keys:
-            del step_in[key]
-
-        if len(step_in) == 0:
-            del step["in"]
-
-    return connect
-
-
 def setup_connected_values(value, key: str = "", append_to: Optional[dict[str, list]] = None) -> Any:
     """Replace links with connected value."""
 

gxformat2/normalized/_format2.py

@@ -290,13 +290,11 @@ def _normalize_outputs(
 
 
 def _pre_clean_steps(workflow: dict[str, Any]) -> dict[str, Any]:
-    """Resolve non-schema connection conventions in step dicts before model validation.
+    """Resolve ``$link`` entries in step state dicts before model validation.
 
-    Handles two conventions not in the Format2 schema:
-    - ``connect`` key on steps → merged into ``in`` as source references
-    - ``$link`` entries in ``state`` → replaced with ConnectedValue, source added to ``in``
-
-    This runs on the raw dict so that model validation sees only schema-compliant data.
+    ``$link`` in ``state`` is a Format2 shorthand for connections embedded in
+    tool state.  This replaces them with ConnectedValue markers and adds the
+    connection source to ``in``, so the model layer sees only schema-compliant data.
     """
     steps = workflow.get("steps", {})
     if isinstance(steps, dict):
@@ -309,44 +307,36 @@ def _pre_clean_steps(workflow: dict[str, Any]) -> dict[str, Any]:
 
 
 def _pre_clean_step(step: dict[str, Any]) -> dict[str, Any]:
-    """Resolve connect and $link on a single step dict."""
-    step = dict(step)
-    in_dict: dict[str, Any] = dict(step.get("in", {})) if isinstance(step.get("in"), dict) else {}
-    in_list: list | None = step.get("in") if isinstance(step.get("in"), list) else None
-    extra_inputs: list[dict[str, Any]] = []
-
-    # Resolve connect key → in entries
-    connect = step.pop("connect", None)
-    if isinstance(connect, dict):
-        for key, sources in connect.items():
-            if in_list is not None:
-                extra_inputs.append({"id": key, "source": sources})
-            else:
-                in_dict[key] = {"source": sources} if isinstance(sources, list) else sources
-
-    # Resolve $link in state → ConnectedValue + in entries
+    """Resolve $link in state on a single step dict."""
     state = step.get("state")
-    if isinstance(state, dict):
-        clean_state, link_connections = _resolve_links(state)
-        step["state"] = clean_state
-        for key, sources in link_connections.items():
-            source = sources if len(sources) > 1 else sources[0]
-            if in_list is not None:
-                extra_inputs.append({"id": key, "source": source})
-            else:
+    if not isinstance(state, dict):
+        # Recursively clean subworkflow runs even if no state
+        run = step.get("run")
+        if isinstance(run, dict) and run.get("class") == "GalaxyWorkflow":
+            return {**step, "run": _pre_clean_steps(run)}
+        return step
+
+    step = dict(step)
+    clean_state, link_connections = _resolve_links(state)
+    step["state"] = clean_state
+
+    if link_connections:
+        in_val = step.get("in")
+        if isinstance(in_val, list):
+            extra = [{"id": k, "source": srcs if len(srcs) > 1 else srcs[0]} for k, srcs in link_connections.items()]
+            step["in"] = in_val + extra
+        else:
+            in_dict = dict(in_val) if isinstance(in_val, dict) else {}
+            for key, sources in link_connections.items():
+                source = sources if len(sources) > 1 else sources[0]
                 in_dict[key] = {"source": source} if isinstance(source, list) else source
+            step["in"] = in_dict
 
     # Recursively clean subworkflow runs
     run = step.get("run")
     if isinstance(run, dict) and run.get("class") == "GalaxyWorkflow":
         step["run"] = _pre_clean_steps(run)
 
-    # Write back in
-    if in_list is not None:
-        step["in"] = in_list + extra_inputs
-    elif in_dict:
-        step["in"] = in_dict
-
     return step
 
 

tests/test_model_helpers.py

@@ -1,49 +1,78 @@
-from gxformat2.model import (
-    pop_connect_from_step_dict,
-    setup_connected_values,
-)
-
-
-def test_pop_connect():
-    raw_step = {
-        "in": {
-            "bar": {
-                "source": "foo/moo",
-            },
-        },
-    }
-    connect = pop_connect_from_step_dict(raw_step)
-    assert connect["bar"] == ["foo/moo"]
-    assert "in" not in raw_step
-
-
-def test_pop_connect_preserves_defaults():
-    raw_step = {
-        "in": {
-            "bar": {
-                "default": 7,
-            },
-        },
-    }
-    connect = pop_connect_from_step_dict(raw_step)
-    assert "bar" not in connect
-    assert "in" in raw_step
-
-
-def test_setup_connected_values():
-    raw_state = {
-        "input": {"$link": "moo/cow"},
-    }
-    connect = {}
-    setup_connected_values(raw_state, append_to=connect)
-    assert connect["input"][0] == "moo/cow"
-
-
-def test_setup_connected_values_in_array():
-    raw_state = {
-        "input": [{"$link": "moo/cow"}, {"$link": "moo/cow2"}],
-    }
-    connect = {}
-    setup_connected_values(raw_state, append_to=connect)
-    assert connect["input"][0] == "moo/cow"
-    assert connect["input"][1] == "moo/cow2"
+"""Tests for connection resolution during normalization."""
+
+from gxformat2.to_format2 import ensure_format2
+from gxformat2.yaml import ordered_load
+
+
+def test_link_in_state_resolved():
+    """$link in state → ConnectedValue in state + source in in_."""
+    nf2 = ensure_format2(ordered_load("""
+class: GalaxyWorkflow
+inputs:
+  moo: data
+steps:
+  s1:
+    tool_id: cat1
+    state:
+      input:
+        $link: moo/cow
+"""))
+    step = nf2.steps[0]
+    assert step.state["input"] == {"__class__": "ConnectedValue"}
+    sources = {si.id: si.source for si in step.in_}
+    assert sources["input"] == "moo/cow"
+
+
+def test_link_array_in_state_resolved():
+    """Multiple $link in array → ConnectedValue markers + sources in in_."""
+    nf2 = ensure_format2(ordered_load("""
+class: GalaxyWorkflow
+inputs:
+  moo: data
+steps:
+  s1:
+    tool_id: cat1
+    state:
+      input:
+        - $link: moo/cow
+        - $link: moo/cow2
+"""))
+    step = nf2.steps[0]
+    assert step.state["input"] == [None, None]
+    sources = {si.id: si.source for si in step.in_}
+    assert "moo/cow" in sources["input"]
+    assert "moo/cow2" in sources["input"]
+
+
+def test_in_source_preserved():
+    """Regular in sources pass through to in_."""
+    nf2 = ensure_format2(ordered_load("""
+class: GalaxyWorkflow
+inputs:
+  foo: data
+steps:
+  s1:
+    tool_id: cat1
+    in:
+      bar: foo/moo
+"""))
+    step = nf2.steps[0]
+    sources = {si.id: si.source for si in step.in_}
+    assert sources["bar"] == "foo/moo"
+
+
+def test_in_default_preserved():
+    """in entries with default (no source) preserved."""
+    nf2 = ensure_format2(ordered_load("""
+class: GalaxyWorkflow
+inputs: {}
+steps:
+  s1:
+    tool_id: cat1
+    in:
+      bar:
+        default: 7
+"""))
+    step = nf2.steps[0]
+    defaults = {si.id: si.default for si in step.in_}
+    assert defaults["bar"] == 7