fix(workflows): namespace nested descendant step ids in loops/fan-out

src/specify_cli/workflows/engine.py

@@ -891,6 +891,76 @@ def append_log(self, entry: dict[str, Any]) -> None:
                 f.write(json.dumps(entry) + "\n")
 
 
+# Nested step keys that may contain a list of steps, mirroring
+# ``overlays/merge.py``'s ``_NESTED_LIST_KEYS`` (this module cannot import
+# that one without a circular import: ``overlays`` imports ``WorkflowDefinition``
+# from here).
+_NESTED_STEP_LIST_KEYS = ("then", "else", "steps", "default")
+
+
+def _rename_step_tree_ids(
+    step: dict[str, Any], prefix: str, suffix: str, *, default_id: str | None = None
+) -> tuple[dict[str, Any], dict[str, str]]:
+    """Return a copy of *step* with every id in its subtree rewritten to
+    ``f"{prefix}:{orig_id}:{suffix}"``, plus a ``{new_id: original_id}`` map.
+
+    A loop iteration or fan-out item previously renamed only the id of the
+    step it iterates over directly (the immediate loop-body/fan-out-template
+    step). A step nested one level deeper — e.g. a ``shell`` step inside an
+    ``if`` inside a ``while`` body or fan-out ``step:`` template — kept its
+    bare, unnamespaced id across every iteration/item, so each iteration/item
+    silently overwrote the previous one's entry in ``context.steps`` /
+    ``state.step_results`` under that same key: only the last iteration's or
+    item's result for that nested step ever survived.
+
+    Recurses into ``then``, ``else``, ``steps``, ``default``, and ``cases.*``
+    — the same nesting keys ``overlays/merge.py`` walks for step-tree
+    attribution — so every descendant gets a unique id, not just the direct
+    child. ``default_id`` supplies the fallback used only when the top-level
+    *step* itself has no ``id`` (mirroring each caller's own historical
+    fallback, e.g. fan-out's ``template.get("id", "item")``); a validated
+    workflow requires an id on every nested step, so nested frames that lack
+    one are left unrenamed rather than guessing a name.
+    """
+    new_step = dict(step)
+    id_map: dict[str, str] = {}
+    orig_id = new_step.get("id") or default_id
+    if isinstance(orig_id, str):
+        new_id = f"{prefix}:{orig_id}:{suffix}"
+        new_step["id"] = new_id
+        id_map[new_id] = orig_id
+    for key in _NESTED_STEP_LIST_KEYS:
+        nested = new_step.get(key)
+        if isinstance(nested, list):
+            renamed_list = []
+            for child in nested:
+                if isinstance(child, dict):
+                    new_child, child_map = _rename_step_tree_ids(child, prefix, suffix)
+                    renamed_list.append(new_child)
+                    id_map.update(child_map)
+                else:
+                    renamed_list.append(child)
+            new_step[key] = renamed_list
+    cases = new_step.get("cases")
+    if isinstance(cases, dict):
+        new_cases = {}
+        for case_key, case_steps in cases.items():
+            if isinstance(case_steps, list):
+                renamed_cases = []
+                for child in case_steps:
+                    if isinstance(child, dict):
+                        new_child, child_map = _rename_step_tree_ids(child, prefix, suffix)
+                        renamed_cases.append(new_child)
+                        id_map.update(child_map)
+                    else:
+                        renamed_cases.append(child)
+                new_cases[case_key] = renamed_cases
+            else:
+                new_cases[case_key] = case_steps
+        new_step["cases"] = new_cases
+    return new_step, id_map
+
+
 # -- Workflow Engine ------------------------------------------------------
 
 
@@ -1346,17 +1416,19 @@ def _execute_steps(
                     for _loop_iter in range(max_iters - 1):
                         if not evaluate_condition(condition, context):
                             break
-                        # Namespace nested step IDs per iteration
-                        # so logs and state keys are unique.
-                        # Execute one step at a time and alias each
-                        # result back to the unprefixed key so that
-                        # later steps in the same body and the loop
-                        # condition see the latest values.
+                        # Namespace nested step IDs (recursively, including
+                        # descendants nested inside e.g. an 'if' in the loop
+                        # body — see _rename_step_tree_ids) per iteration so
+                        # logs and state keys are unique. Execute one step at
+                        # a time and alias each renamed id in the subtree back
+                        # to its original, unprefixed id so that later steps
+                        # in the same body and the loop condition see the
+                        # latest values.
                         for ns_idx, ns in enumerate(result.next_steps):
-                            ns_copy = dict(ns)
-                            orig = ns_copy.get("id")
-                            base_id = orig or f"step-{ns_idx}"
-                            ns_copy["id"] = f"{step_id}:{base_id}:{_loop_iter + 1}"
+                            ns_copy, id_map = _rename_step_tree_ids(
+                                ns, step_id, str(_loop_iter + 1),
+                                default_id=f"step-{ns_idx}",
+                            )
                             self._execute_steps(
                                 [ns_copy], context, state, registry,
                                 step_offset=-1,
@@ -1367,11 +1439,12 @@ def _execute_steps(
                                 RunStatus.ABORTED,
                             ):
                                 return
-                            if orig and ns_copy["id"] in context.steps:
-                                self._record_result(
-                                    context, state, orig,
-                                    context.steps[ns_copy["id"]],
-                                )
+                            for new_id, orig_id in id_map.items():
+                                if new_id in context.steps:
+                                    self._record_result(
+                                        context, state, orig_id,
+                                        context.steps[new_id],
+                                    )
 
             # Fan-out: execute the nested step template once per item. Honors
             # max_concurrency — <=1 runs sequentially (default, historical
@@ -1458,11 +1531,28 @@ def item_id(idx: int) -> str:
             return f"{step_id}:{base_id}:{idx}"
 
         def run_item(idx: int, item_ctx: StepContext) -> Any:
-            item_step = dict(template)
-            item_step["id"] = item_id(idx)
+            # Namespace every id in the template's subtree (not just the
+            # template's own top-level id) so a step nested inside e.g. an
+            # 'if'/'switch' branch of the fan-out template gets a unique key
+            # per item instead of colliding across items — and, more
+            # seriously, potentially colliding with an unrelated step of the
+            # same id elsewhere in the workflow (fan-out templates are
+            # exempted from the global id-uniqueness check specifically
+            # because runtime namespacing was assumed to make collisions
+            # safe; see _rename_step_tree_ids). Each renamed descendant is
+            # then aliased back to its original id so sibling steps within
+            # the same item's template and code reading `steps.<id>.output`
+            # after the fan-out still see that item's value (mirroring the
+            # while/do-while loop body's behavior).
+            item_step, id_map = _rename_step_tree_ids(
+                template, step_id, str(idx), default_id=base_id,
+            )
             self._execute_steps(
                 [item_step], item_ctx, state, registry, step_offset=-1,
             )
+            for new_id, orig_id in id_map.items():
+                if new_id in item_ctx.steps:
+                    self._record_result(item_ctx, state, orig_id, item_ctx.steps[new_id])
             # Read back through the context that was actually executed against,
             # not the outer closure — clearer and robust if StepContext copying
             # ever stops sharing the steps dict by reference.

tests/test_workflows.py

@@ -6298,6 +6298,100 @@ def test_loop_with_bool_max_iterations_uses_default_cap(self, project_dir):
         # Falls back to the default cap of 10, not range(True - 1) == 1 run.
         assert counter_file.read_text(encoding="utf-8").strip() == "10"
 
+    def test_while_loop_namespaces_nested_descendant_steps(self, project_dir):
+        """A step nested one level deeper than the loop body's direct child
+        (e.g. a `shell` step inside an `if` inside the `while` body) must get
+        a unique namespaced key per iteration, not just the immediate child.
+
+        Previously only the direct child's id was namespaced
+        (`retry-loop:guard:1`); the grandchild `leaf` kept its bare id across
+        every iteration, so each iteration silently overwrote the previous
+        one's entry in `state.step_results["leaf"]` and no per-iteration
+        record of it ever existed.
+        """
+        from specify_cli.workflows.engine import WorkflowEngine, WorkflowDefinition
+        from specify_cli.workflows.base import RunStatus
+
+        yaml_str = """
+schema_version: "1.0"
+workflow:
+  id: "while-nested-descendant"
+  name: "While Nested Descendant"
+  version: "1.0.0"
+steps:
+  - id: retry-loop
+    type: while
+    condition: "true"
+    max_iterations: 3
+    steps:
+      - id: guard
+        type: if
+        condition: "true"
+        then:
+          - id: leaf
+            type: shell
+            run: "echo tick"
+"""
+        definition = WorkflowDefinition.from_string(yaml_str)
+        engine = WorkflowEngine(project_dir)
+        state = engine.execute(definition)
+
+        assert state.status == RunStatus.COMPLETED
+        # The unprefixed key still holds the latest iteration's result
+        # (sibling steps in the loop body and the loop condition read it).
+        assert state.step_results["leaf"]["output"]["stdout"] == "tick\n"
+        # Every iteration's grandchild result is separately recoverable.
+        assert "retry-loop:leaf:1" in state.step_results
+        assert "retry-loop:leaf:2" in state.step_results
+
+    def test_fan_out_namespaces_nested_descendant_steps(self, project_dir):
+        """A step nested inside a fan-out template's `if`/`switch` branch
+        must get a unique namespaced key per item, not just the template's
+        own top-level id.
+
+        Previously only the template's own id was namespaced
+        (`fan:item:0`); a grandchild step like `leaf` kept its bare id
+        across every item, so each item silently overwrote the previous
+        item's entry in `state.step_results["leaf"]` — losing every item's
+        nested result except the last. Nested/template step ids are exempt
+        from the workflow's global id-uniqueness validation specifically
+        because runtime namespacing is assumed to make collisions safe, so
+        an unnamespaced grandchild id can also collide with an unrelated
+        step of the same id elsewhere in the workflow.
+        """
+        from specify_cli.workflows.engine import WorkflowEngine, WorkflowDefinition
+        from specify_cli.workflows.base import RunStatus
+
+        yaml_str = """
+schema_version: "1.0"
+workflow:
+  id: "fan-out-nested-descendant"
+  name: "Fan Out Nested Descendant"
+  version: "1.0.0"
+steps:
+  - id: fan
+    type: fan-out
+    items: "{{ ['a', 'b', 'c'] }}"
+    max_concurrency: 1
+    step:
+      id: item
+      type: if
+      condition: "true"
+      then:
+        - id: leaf
+          type: shell
+          run: "echo {{ item }}"
+"""
+        definition = WorkflowDefinition.from_string(yaml_str)
+        engine = WorkflowEngine(project_dir)
+        state = engine.execute(definition)
+
+        assert state.status == RunStatus.COMPLETED
+        # Every item's grandchild result is separately recoverable.
+        assert state.step_results["fan:leaf:0"]["output"]["stdout"] == "a\n"
+        assert state.step_results["fan:leaf:1"]["output"]["stdout"] == "b\n"
+        assert state.step_results["fan:leaf:2"]["output"]["stdout"] == "c\n"
+
     def test_do_while_loop_runs_to_max_when_condition_stays_true(self, project_dir):
         """Do-while loop must still run to max_iterations when the condition
         never becomes false.