Fix a bug in pipeweld for inserting into parallel steps with empty … (#1419)

nathanjmcdougall · web-flow · commit 889eb18fe11e · 2026-03-14T09:22:52.000+13:00
* Fix a bug in `pipeweld` for inserting into parallel steps with empty series

Enable `reportUnknownLambdaType` in `basedpyright`

* Handle bug with a parallelism containing a multi-step series component

* Avoid adding a successor into an empty parallelism

* Fix empty components bug
diff --git a/pyproject.toml b/pyproject.toml
@@ -222,14 +222,14 @@ reportMissingTypeStubs = false
 reportPrivateUsage = false
 reportUnannotatedClassAttribute = false
 reportUnknownArgumentType = false
-reportUnknownLambdaType = false
 reportUnknownMemberType = false
 reportUnknownVariableType = false
 reportUnnecessaryIsInstance = false
 reportUnusedCallResult = false
 
 [[tool.basedpyright.executionEnvironments]]
 root = "tests"
+reportUnknownLambdaType = false
 reportUnknownParameterType = false
 reportUnreachable = false
 reportUnusedFunction = false
diff --git a/src/usethis/_pipeweld/func.py b/src/usethis/_pipeweld/func.py
@@ -465,7 +465,8 @@ def _get_instructions_for_insertion(
 
     Args:
         component: The component to insert.
-        after: The step to insert the new component after.
+        after: The step to insert the new component after. None if the component should
+               be inserted at the start of the pipeline.
 
     Returns:
         A tuple containing the instructions to insert the new component and the endpoint
@@ -486,40 +487,51 @@ def _get_instructions_for_insertion(
         if len(component.root) == 0:
             return [], after
 
-        instructions: list[Instruction] = []
-        endpoints = []
-        min_idx = None
-        min_endpoint = None
+        instructions_per_sub: list[list[Instruction]] = []
+        endpoints: list[str | None] = []
+        min_idx: int | None = None
+        min_endpoint: str | None = None
         for idx, subcomponent in enumerate(component.root):
             new_instructions, endpoint = _get_instructions_for_insertion(
                 subcomponent,
                 after=after,
             )
-            if endpoint is not None and (
-                min_endpoint is None or endpoint < min_endpoint
+            if (
+                new_instructions
+                and endpoint is not None
+                and (min_endpoint is None or endpoint < min_endpoint)
             ):
                 min_idx = idx
                 min_endpoint = endpoint
 
             endpoints.append(endpoint)
-            instructions.extend(new_instructions)
+            instructions_per_sub.append(new_instructions)
+        if min_endpoint is None:
+            # If the existing components are empty...
+            min_endpoint = after
 
         for idx in range(len(component.root)):
-            if idx != min_idx:
-                instructions[idx] = InsertParallel(
-                    after=instructions[idx].after, step=instructions[idx].step
+            if idx != min_idx and instructions_per_sub[idx]:
+                instructions_per_sub[idx][0] = InsertParallel(
+                    after=instructions_per_sub[idx][0].after,
+                    step=instructions_per_sub[idx][0].step,
                 )
 
-        sorted_idxs = sorted(range(len(endpoints)), key=lambda k: endpoints[k])
+        sorted_idxs = sorted(
+            range(len(endpoints)),
+            key=lambda k: (
+                endpoints[k] is None,
+                endpoints[k] if endpoints[k] is not None else "",
+            ),
+        )
 
-        instructions = [instructions[idx] for idx in sorted_idxs]
+        instructions = [
+            inst for idx in sorted_idxs for inst in instructions_per_sub[idx]
+        ]
 
-        return instructions, min(endpoints)
+        return instructions, min_endpoint
     elif isinstance(component, DepGroup):
-        return _get_instructions_for_insertion(
-            component.series,
-            after=after,
-        )
+        return _get_instructions_for_insertion(component.series, after=after)
     else:
         assert_never(component)
 
diff --git a/src/usethis/_tool/impl/base/import_linter.py b/src/usethis/_tool/impl/base/import_linter.py
@@ -43,7 +43,7 @@ def config_spec(self) -> ConfigSpec:
             default=0,
         )
 
-        contracts: list[dict] = []
+        contracts: list[dict[str, bool | str | list[str]]] = []
         for (
             layered_architecture_by_module
         ) in layered_architecture_by_module_by_root_package.values():
@@ -63,7 +63,7 @@ def config_spec(self) -> ConfigSpec:
                 ):
                     continue
 
-                layers = []
+                layers: list[str] = []
                 for layer in layered_architecture.layers:
                     layers.append(" | ".join(sorted(layer)))
 
diff --git a/tests/usethis/_integrations/ci/bitbucket/test_pipeweld.py b/tests/usethis/_integrations/ci/bitbucket/test_pipeweld.py
@@ -726,6 +726,70 @@ def test_multiple(self):
             ]
             assert endpoint == "A"  # Alphabetical order wins tiebreaks
 
+        def test_contains_empty_series(self):
+            """A Parallel with an empty Series subcomponent produces a None endpoint
+            for that branch; the overall endpoint should be the min non-None endpoint."""
+            # Arrange
+            component = parallel("B", series())
+            after = None
+
+            # Act
+            instructions, endpoint = _get_instructions_for_insertion(
+                component, after=after
+            )
+
+            # Assert
+            assert instructions == [InsertSuccessor(step="B", after=None)]
+            assert endpoint == "B"
+
+        def test_contains_multi_step_series(self):
+            # Arrange
+            component = parallel("A", series("B", "C"))
+            after = "0"
+
+            # Act
+            instructions, endpoint = _get_instructions_for_insertion(
+                component, after=after
+            )
+
+            # Assert
+            assert instructions == [
+                InsertSuccessor(step="A", after="0"),
+                InsertParallel(step="B", after="0"),
+                InsertSuccessor(step="C", after="B"),
+            ]
+            assert endpoint == "A"
+
+        def test_empty_series_with_smaller_after(self):
+            # Arrange
+            component = parallel("C", series())
+            after = "A"
+
+            # Act
+            instructions, endpoint = _get_instructions_for_insertion(
+                component, after=after
+            )
+
+            # Assert
+            assert instructions == [InsertSuccessor(step="C", after="A")]
+            assert endpoint == "C"
+
+        def test_all_empty_branches_preserves_after(self):
+            """When all branches are empty, no instructions are produced but the
+            endpoint should still be the original ``after`` value."""
+            # Arrange
+            component = parallel(series(), series())
+            after = "X"
+
+            # Act
+            instructions, endpoint = _get_instructions_for_insertion(
+                component, after=after
+            )
+
+            # Assert
+            assert instructions == []
+            assert endpoint == "X"
+
     class TestDepGroup:
         def test_basic(self):
             # Arrange
