From d6423c1b8eb2bbbae0a7c0336c8e08c594680011 Mon Sep 17 00:00:00 2001
From: Noor-ul-ain001 <noor01mk@gmail.com>
Date: Wed, 26 Aug 2026 15:35:49 +0500
Subject: [PATCH 1/4] fix(workflows): namespace nested descendant step ids in
 loops/fan-out
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

`while`/`do-while` loop bodies and `fan-out` templates namespace nested
step ids per iteration/item so logs and `state.step_results` entries stay
unique — but the namespacing only rewrote the id of the *immediate* child
step, not any descendant nested deeper (e.g. a `shell` step inside an `if`
inside a `while` body, or inside a `fan-out` template's `if`/`switch`
branch). That grandchild kept its bare, unnamespaced id across every
iteration/item, so each iteration/item silently overwrote the previous
one's entry in `state.step_results` under that same key — only the last
iteration's or item's result for that nested step ever survived, and no
per-iteration/per-item record of it ever existed.

This is also a correctness gap beyond bookkeeping: nested/template step ids
are deliberately exempted from the workflow's global id-uniqueness
validation, on the assumption that runtime namespacing makes any collision
safe. Since only the top-level child was actually namespaced, a step
nested one level deeper could collide with an unrelated step of the same
id elsewhere in the workflow and silently overwrite its result.

Fix: add `_rename_step_tree_ids`, which recursively rewrites every id in a
step's subtree (walking `then`/`else`/`steps`/`default`/`cases.*` — the
same nesting keys `overlays/merge.py` walks for step-tree attribution) and
returns a `{new_id: original_id}` map. Both the while/do-while loop body
and fan-out's `run_item` now use this helper instead of renaming only the
top-level id, and alias every renamed descendant's result back to its
original id (mirroring the existing single-level aliasing) so sibling
steps within the same iteration/item and code reading `steps.<id>.output`
after the loop/fan-out still see that iteration's/item's value.

## Test plan
- Added `test_while_loop_namespaces_nested_descendant_steps` and
  `test_fan_out_namespaces_nested_descendant_steps` to
  `tests/test_workflows.py::TestWorkflowEngine`: a `shell` step nested
  inside an `if` inside a `while` body (and inside a `fan-out` template)
  gets a distinct namespaced `state.step_results` entry per
  iteration/item, while the unprefixed key still holds the latest value.
- Verified both fail without the fix (test-the-test): the namespaced keys
  (`retry-loop:leaf:1`, `fan:leaf:0`, etc.) were simply absent, and
  `step_results` only ever held the last iteration's/item's bare-keyed
  entry — reproducing the exact bug.
- Ran the full `tests/test_workflows.py` suite: 926 passed, 20 pre-existing
  Windows symlink-elevation failures (need admin rights, unrelated to this
  change), 7 skipped. All `While`/`DoWhile`/`FanOut`/`FanOutConcurrency`
  tests pass, including the concurrent-execution and per-thread context
  isolation tests.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PJHJ2dHP2RVCNncHqN8Qm9
---
 src/specify_cli/workflows/engine.py | 124 ++++++++++++++++++++++++----
 tests/test_workflows.py             |  94 +++++++++++++++++++++
 2 files changed, 201 insertions(+), 17 deletions(-)

diff --git a/src/specify_cli/workflows/engine.py b/src/specify_cli/workflows/engine.py
index d17513cc0b..8ff83da56a 100644
--- a/src/specify_cli/workflows/engine.py
+++ b/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.
diff --git a/tests/test_workflows.py b/tests/test_workflows.py
index d599f3c6a4..c91f31ea8c 100644
--- a/tests/test_workflows.py
+++ b/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.

From c1d7eca9034b16a6b00e7aac8542310ee5c8ed85 Mon Sep 17 00:00:00 2001
From: Noor-ul-ain001 <noor01mk@gmail.com>
Date: Thu, 3 Sep 2026 01:01:16 +0500
Subject: [PATCH 2/4] fix(workflows): namespace loop iteration 0 and alias
 immediately, not post-subtree

Addresses Copilot review feedback on PR #4338:

- while/do-while loop iteration 0 ran through a separate, unnamespaced
  code path before the loop-specific namespacing logic was reached, so
  it had no dedicated state.step_results entry and was immediately
  overwritten the moment iteration 1's aliasing ran. Every iteration,
  including the first, now goes through the same
  _rename_step_tree_ids + alias_map path.

- Bare-id aliasing for a namespaced descendant happened only after its
  entire renamed subtree finished executing, so a later sibling step
  in the same iteration/item that referenced an earlier sibling by its
  original id ran before that alias existed and read a stale (or
  absent) value. _execute_steps now threads an alias_map through so
  each descendant is aliased immediately after it completes, not in
  bulk afterward.

- For a concurrent fan-out (max_concurrency > 1), that same bare-id
  alias write raced across worker threads sharing context.steps, so
  one item's sibling read could observe another item's value. Each
  concurrent item now runs against a private ChainMap overlay for its
  bare-id aliases; only the namespaced (disjoint-key) result is
  published to shared state during execution. Once every item has
  finished (back on the single thread), the last item's aliases are
  applied to shared state once, deterministically.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NhR6g8xT8at5pPMhkrC3e2
---
 src/specify_cli/workflows/engine.py | 188 ++++++++++++++++++++--------
 tests/test_workflows.py             | 147 +++++++++++++++++++++-
 2 files changed, 283 insertions(+), 52 deletions(-)

diff --git a/src/specify_cli/workflows/engine.py b/src/specify_cli/workflows/engine.py
index 8ff83da56a..a50a676f3c 100644
--- a/src/specify_cli/workflows/engine.py
+++ b/src/specify_cli/workflows/engine.py
@@ -17,6 +17,7 @@
 import tempfile
 import threading
 import uuid
+from collections import ChainMap
 from concurrent.futures import Future, ThreadPoolExecutor
 from datetime import datetime, timezone
 from pathlib import Path
@@ -1245,8 +1246,27 @@ def _execute_steps(
         registry: dict[str, Any],
         *,
         step_offset: int = 0,
+        alias_map: dict[str, str] | None = None,
+        alias_local_only: bool = False,
     ) -> None:
-        """Execute a list of steps sequentially."""
+        """Execute a list of steps sequentially.
+
+        ``alias_map`` (``{namespaced_id: original_id}``, from
+        ``_rename_step_tree_ids``) mirrors each recorded step's result to its
+        original, unprefixed id *immediately* after that step finishes -- not
+        after its whole subtree finishes -- so a sibling step later in the
+        same loop iteration / fan-out item that references an earlier
+        sibling by its original id (``steps.<id>``) sees that value right
+        away. It is propagated through the recursive nested-step call below
+        so descendants nested arbitrarily deep (e.g. an ``if`` inside the
+        namespaced step) are aliased too, not just the immediate child.
+
+        ``alias_local_only`` routes that mirror through ``context.steps``
+        only, never ``state.step_results``. A concurrent fan-out item passes
+        a private overlay as ``context.steps`` and this flag so concurrently
+        running items never race to write the same original id in shared
+        state; see ``_run_fan_out``.
+        """
         for i, step_config in enumerate(steps):
             step_id = step_config.get("id", f"step-{i}")
             step_type = step_config.get("type", "command")
@@ -1301,6 +1321,13 @@ def _execute_steps(
                 "error": result.error,
             }
             self._record_result(context, state, step_id, step_data)
+            if alias_map is not None:
+                orig_id = alias_map.get(step_id)
+                if orig_id is not None:
+                    if alias_local_only:
+                        context.steps[orig_id] = step_data
+                    else:
+                        self._record_result(context, state, orig_id, step_data)
 
             state.append_log(
                 {
@@ -1385,18 +1412,14 @@ def _execute_steps(
             # A step-path stack for exact nested resume is a future
             # enhancement.
             if result.next_steps:
-                self._execute_steps(
-                    result.next_steps, context, state, registry,
-                    step_offset=-1,
-                )
-                if state.status in (
-                    RunStatus.PAUSED,
-                    RunStatus.FAILED,
-                    RunStatus.ABORTED,
-                ):
-                    return
-
-                # Loop iteration: while/do-while re-evaluate after body
+                # Loop iteration: while/do-while re-evaluate after body. Every
+                # iteration -- including the first -- is namespaced and run
+                # through the same _rename_step_tree_ids + alias_map path, so
+                # each has its own state.step_results entry (see
+                # _rename_step_tree_ids). Previously only iterations after the
+                # first were namespaced: the first ran with bare ids and no
+                # dedicated entry, and iteration 1's aliasing then silently
+                # overwrote it, making iteration 0's result unrecoverable.
                 if step_type in ("while", "do-while"):
                     from .expressions import evaluate_condition
 
@@ -1413,25 +1436,30 @@ def _execute_steps(
                     ):
                         max_iters = 10
                     condition = step_config.get("condition", False)
-                    for _loop_iter in range(max_iters - 1):
-                        if not evaluate_condition(condition, context):
+                    for _loop_iter in range(max_iters):
+                        if _loop_iter > 0 and not evaluate_condition(
+                            condition, context
+                        ):
                             break
                         # 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.
+                        # a time; alias_map aliases each renamed id in the
+                        # subtree back to its original, unprefixed id
+                        # immediately as that step completes (not after the
+                        # whole iteration finishes), so 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, id_map = _rename_step_tree_ids(
-                                ns, step_id, str(_loop_iter + 1),
+                                ns, step_id, str(_loop_iter),
                                 default_id=f"step-{ns_idx}",
                             )
                             self._execute_steps(
                                 [ns_copy], context, state, registry,
-                                step_offset=-1,
+                                step_offset=-1, alias_map=id_map,
+                                alias_local_only=alias_local_only,
                             )
                             if state.status in (
                                 RunStatus.PAUSED,
@@ -1439,12 +1467,18 @@ def _execute_steps(
                                 RunStatus.ABORTED,
                             ):
                                 return
-                            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],
-                                    )
+                else:
+                    self._execute_steps(
+                        result.next_steps, context, state, registry,
+                        step_offset=-1, alias_map=alias_map,
+                        alias_local_only=alias_local_only,
+                    )
+                    if state.status in (
+                        RunStatus.PAUSED,
+                        RunStatus.FAILED,
+                        RunStatus.ABORTED,
+                    ):
+                        return
 
             # Fan-out: execute the nested step template once per item. Honors
             # max_concurrency — <=1 runs sequentially (default, historical
@@ -1530,7 +1564,9 @@ def item_id(idx: int) -> str:
             # Per-item ID grammar: parentId:templateId:index.
             return f"{step_id}:{base_id}:{idx}"
 
-        def run_item(idx: int, item_ctx: StepContext) -> Any:
+        def run_item(
+            idx: int, item_ctx: StepContext, *, local_only: bool
+        ) -> tuple[Any, dict[str, dict[str, Any]]]:
             # 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
@@ -1539,26 +1575,54 @@ def run_item(idx: int, item_ctx: StepContext) -> Any:
             # 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).
+            # safe; see _rename_step_tree_ids).
             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,
-            )
+            # ``local_only`` (concurrent path): give this item a private
+            # overlay for its ``.steps`` reads/writes. Namespaced results
+            # still land in the real ``state.step_results`` (via
+            # _record_result's unconditional write — see _execute_steps),
+            # but the immediate bare-id alias (see alias_map below) writes
+            # only into this overlay. That lets a later sibling step in
+            # THIS item's template resolve an earlier sibling by its
+            # original id via the overlay, without ever mutating the
+            # shared steps dict that other concurrently-running items also
+            # read from — the actual race Copilot flagged: every worker
+            # writing the same bare-id key could otherwise expose another
+            # item's value to a sibling read. The caller applies exactly
+            # one item's aliases to shared state — deterministically the
+            # last item in item order — once every item has finished.
+            original_steps = item_ctx.steps
+            local_overlay: dict[str, dict[str, Any]] = {}
+            if local_only:
+                item_ctx.steps = ChainMap(local_overlay, original_steps)
+            try:
+                self._execute_steps(
+                    [item_step], item_ctx, state, registry, step_offset=-1,
+                    alias_map=id_map, alias_local_only=local_only,
+                )
+            finally:
+                item_ctx.steps = original_steps
+            alias_records: dict[str, dict[str, Any]] = {}
+            steps_view = local_overlay if local_only else item_ctx.steps
             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.
-            return item_ctx.steps.get(item_step["id"], {}).get("output", {})
-
-        # Sequential path — identical to the historical behavior.
+                if new_id in steps_view:
+                    data = steps_view[new_id]
+                    if local_only:
+                        # Publish the namespaced (disjoint, per-item) result
+                        # into the truly-shared steps dict explicitly — safe
+                        # even under concurrency since each item only ever
+                        # writes its own namespaced keys here.
+                        original_steps[new_id] = data
+                    alias_records[orig_id] = data
+            # Read back through the local view, not the outer closure —
+            # clearer and robust if StepContext copying ever stops sharing
+            # the steps dict by reference.
+            return steps_view.get(item_step["id"], {}).get("output", {}), alias_records
+
+        # Sequential path — identical to the historical behavior, plus
+        # immediate (not post-subtree) bare-id aliasing.
         if workers <= 1:
             results: list[Any] = []
             previous_item = context.item
@@ -1567,7 +1631,10 @@ def run_item(idx: int, item_ctx: StepContext) -> Any:
             try:
                 for item_idx, item_val in enumerate(items):
                     context.item = item_val
-                    results.append(run_item(item_idx, context))
+                    output, _alias_records = run_item(
+                        item_idx, context, local_only=False
+                    )
+                    results.append(output)
                     if state.status in halting:
                         break
             finally:
@@ -1578,11 +1645,13 @@ def run_item(idx: int, item_ctx: StepContext) -> Any:
         # Concurrent path — bounded sliding window; results assembled in item order.
         n = len(items)
         slots: list[Any] = [None] * n
+        alias_slots: list[dict[str, dict[str, Any]]] = [{}] * n
 
-        def run_isolated(idx: int) -> Any:
+        def run_isolated(idx: int) -> tuple[Any, dict[str, dict[str, Any]]]:
             # Each item runs against its own context copy so context.item is not
-            # clobbered across threads; the shared steps dict is written only on the
-            # disjoint parentId:templateId:index key (GIL-safe on distinct keys).
+            # clobbered across threads; local_only=True gives it a private
+            # steps overlay so its immediate bare-id aliases cannot race a
+            # concurrently-running sibling item's aliases (see run_item).
             return run_item(
                 idx,
                 dataclasses.replace(
@@ -1590,6 +1659,7 @@ def run_isolated(idx: int) -> Any:
                     item=items[idx],
                     inside_fan_out=True,
                 ),
+                local_only=True,
             )
 
         def item_halt_status(idx: int) -> RunStatus | None:
@@ -1599,7 +1669,13 @@ def item_halt_status(idx: int) -> RunStatus | None:
             # misattributed here. Mirrors the sequential mapping: PAUSED -> PAUSED;
             # FAILED -> ABORTED when aborted, else FAILED, unless continue_on_error
             # routes around it.
-            rec = context.steps.get(item_id(idx))
+            # Reads from state.step_results (not context.steps): a concurrent
+            # item's steps overlay is private (see run_item), so
+            # context.steps is no longer guaranteed to carry this item's
+            # namespaced entry, while state.step_results always does — every
+            # namespaced write reaches it unconditionally regardless of the
+            # overlay.
+            rec = state.step_results.get(item_id(idx))
             if rec is None:
                 # Ran but recorded nothing — only when the item failed before
                 # record_step_result (e.g. an unknown step type returns early).
@@ -1644,7 +1720,7 @@ def item_halt_status(idx: int) -> RunStatus | None:
                     # change ever breaks that invariant.
                     break
                 try:
-                    slots[idx] = fut.result()
+                    slots[idx], alias_slots[idx] = fut.result()
                 except Exception:
                     # A genuine exception escaping a step (not a normal step
                     # FAILED, which sets state.status) must not be masked: cancel
@@ -1665,6 +1741,16 @@ def item_halt_status(idx: int) -> RunStatus | None:
                         other.cancel()
                     break
 
+        # Apply exactly one item's bare-id aliases to the real shared state —
+        # deterministically the last item in item order (the halting item, if
+        # any, else the last one collected) — now that the pool has joined
+        # and this runs single-threaded again, so it can never race a
+        # concurrently-running item the way writing it during run_item would.
+        last_idx = halt[0] if halt is not None else (collected - 1 if collected else None)
+        if last_idx is not None:
+            for orig_id, data in alias_slots[last_idx].items():
+                self._record_result(context, state, orig_id, data)
+
         if halt is not None:
             halted_at, halted_status = halt
             # A later in-flight item may have overwritten state.status before the
@@ -1678,7 +1764,7 @@ def item_halt_status(idx: int) -> RunStatus | None:
             # third-party step returning FAILED with no message never inherits
             # an unrelated concurrent item's error; this mirrors the sequential
             # path, which sets state.error = result.error verbatim.
-            halt_rec = context.steps.get(item_id(halted_at))
+            halt_rec = state.step_results.get(item_id(halted_at))
             if isinstance(halt_rec, dict):
                 state.error = halt_rec.get("error")
             return slots[: halted_at + 1]
diff --git a/tests/test_workflows.py b/tests/test_workflows.py
index c91f31ea8c..8c9135419f 100644
--- a/tests/test_workflows.py
+++ b/tests/test_workflows.py
@@ -6340,10 +6340,155 @@ def test_while_loop_namespaces_nested_descendant_steps(self, project_dir):
         # 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.
+        # Every iteration's grandchild result is separately recoverable --
+        # including the FIRST iteration. The first iteration previously ran
+        # through a separate, unnamespaced code path before the loop-specific
+        # namespacing logic was reached, so it had no dedicated entry and was
+        # immediately overwritten by iteration 1's aliasing the moment that
+        # iteration ran.
+        assert "retry-loop:leaf:0" in state.step_results
         assert "retry-loop:leaf:1" in state.step_results
         assert "retry-loop:leaf:2" in state.step_results
 
+    def test_while_loop_sibling_step_sees_immediate_alias(self, tmp_path):
+        """A step nested inside an `if` in a `while` body that references an
+        earlier SIBLING nested in the SAME `if` branch by its bare id must
+        see that sibling's value from the SAME iteration -- not a stale
+        value left over from a previous iteration.
+
+        Aliasing a namespaced descendant back to its bare id previously
+        happened only after the entire renamed subtree (here, the whole
+        `if` step, both its own id and its branch's) finished executing --
+        so a later sibling in the same branch that read the earlier one by
+        its bare id ran before that iteration's alias was ever written, and
+        so saw the previous iteration's aliased value instead.
+        """
+        from specify_cli.workflows.base import (
+            RunStatus,
+            StepBase,
+            StepContext,
+            StepResult,
+            StepStatus,
+        )
+        from specify_cli.workflows.engine import RunState, WorkflowEngine
+        from specify_cli.workflows.steps.if_then import IfThenStep
+        from specify_cli.workflows.steps.while_loop import WhileStep
+
+        call_count = {"n": 0}
+
+        class _WriteStep(StepBase):
+            type_key = "write"
+
+            def execute(self, config, context):
+                n = call_count["n"]
+                call_count["n"] += 1
+                return StepResult(
+                    status=StepStatus.COMPLETED, output={"marker": f"value-{n}"}
+                )
+
+        class _ReadStep(StepBase):
+            type_key = "read"
+
+            def execute(self, config, context):
+                seen = context.steps.get("first", {}).get("output", {}).get("marker")
+                return StepResult(status=StepStatus.COMPLETED, output={"seen": seen})
+
+        engine = WorkflowEngine(project_root=tmp_path)
+        context = StepContext()
+        state = RunState(run_id="r", workflow_id="w", project_root=tmp_path)
+        state.status = RunStatus.RUNNING
+        registry = {
+            "while": WhileStep(),
+            "if": IfThenStep(),
+            "write": _WriteStep(),
+            "read": _ReadStep(),
+        }
+        steps = [
+            {
+                "id": "retry-loop",
+                "type": "while",
+                "condition": "true",
+                "max_iterations": 2,
+                "steps": [
+                    {
+                        "id": "guard",
+                        "type": "if",
+                        "condition": "true",
+                        "then": [
+                            {"id": "first", "type": "write"},
+                            {"id": "second", "type": "read"},
+                        ],
+                    },
+                ],
+            },
+        ]
+        engine._execute_steps(steps, context, state, registry)
+
+        assert state.status == RunStatus.RUNNING
+        assert state.step_results["retry-loop:second:0"]["output"]["seen"] == "value-0"
+        assert state.step_results["retry-loop:second:1"]["output"]["seen"] == "value-1"
+
+    def test_fan_out_concurrent_sibling_step_isolated_per_item(self, tmp_path):
+        """A later sibling step in a CONCURRENT fan-out item's template that
+        references an earlier sibling by its bare id must see THIS item's
+        value -- not a stale value, and not a value written by a DIFFERENT,
+        concurrently-running item through the same shared bare-id key.
+
+        A barrier forces every item's first sibling to complete at roughly
+        the same time, maximizing the window for a racy implementation to
+        leak one item's value onto another's read of the shared bare-id key.
+        """
+        import threading
+
+        from specify_cli.workflows.base import (
+            RunStatus,
+            StepBase,
+            StepContext,
+            StepResult,
+            StepStatus,
+        )
+        from specify_cli.workflows.engine import RunState, WorkflowEngine
+        from specify_cli.workflows.steps.if_then import IfThenStep
+
+        n = 4
+        barrier = threading.Barrier(n, timeout=5)
+
+        class _WriteStep(StepBase):
+            type_key = "write"
+
+            def execute(self, config, context):
+                barrier.wait()
+                return StepResult(
+                    status=StepStatus.COMPLETED, output={"marker": context.item}
+                )
+
+        class _ReadStep(StepBase):
+            type_key = "read"
+
+            def execute(self, config, context):
+                seen = context.steps.get("first", {}).get("output", {}).get("marker")
+                return StepResult(status=StepStatus.COMPLETED, output={"seen": seen})
+
+        engine = WorkflowEngine(project_root=tmp_path)
+        context = StepContext()
+        state = RunState(run_id="r", workflow_id="w", project_root=tmp_path)
+        state.status = RunStatus.RUNNING
+        registry = {"if": IfThenStep(), "write": _WriteStep(), "read": _ReadStep()}
+        template = {
+            "id": "item",
+            "type": "if",
+            "condition": "true",
+            "then": [
+                {"id": "first", "type": "write"},
+                {"id": "second", "type": "read"},
+            ],
+        }
+        items = list(range(n))
+        engine._run_fan_out(items, template, "fan", context, state, registry, n)
+
+        for i in items:
+            assert state.step_results[f"fan:second:{i}"]["output"]["seen"] == i
+
     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

From 7365250ca69bfda74751161dbfdf6d418fd8aa6e Mon Sep 17 00:00:00 2001
From: Noor-ul-ain001 <noor01mk@gmail.com>
Date: Sun, 13 Sep 2026 20:15:00 +0500
Subject: [PATCH 3/4] fix(workflows): stop double-renaming loop bodies and
 unsafe fan-out aliasing

Two follow-up bugs in the nested step-id namespacing added for loops and
fan-out templates:

1. A while/do-while step nested inside an outer loop iteration or
   fan-out item had its OWN 'steps' body eagerly renamed by the outer
   _rename_step_tree_ids pass (since 'steps' is a walked nesting key).
   When that while step then ran its own per-iteration rename, it treated
   the already-namespaced id (e.g. "fan:leaf:0") as the original and
   aliased back to that synthetic id instead of the workflow author's
   real bare id ("leaf") -- so a doubly-prefixed id like
   "fan:while:0:fan:leaf:0:0" was the only place the value ever landed,
   and state.step_results["leaf"] was never populated at all.
   _rename_step_tree_ids now still renames a nested while/do-while step's
   own id, but leaves its 'steps' body untouched for the loop's own
   runtime namespacing to rename exactly once, against the real ids.

2. A fan-out template's bare-id convenience alias (writing the current
   item's result under its unprefixed id, so `steps.<id>` sees the latest
   value) could silently clobber an unrelated, distinctly-authored step's
   result if the two happened to share an id -- fan-out templates are
   deliberately exempt from the workflow's global id-uniqueness check, so
   nothing prevents that collision. This applied to both the sequential
   path's immediate alias write and the concurrent path's deferred
   post-join write. Added _collect_reserved_step_ids, which computes the
   set of step ids declared outside any fan-out template (mirroring
   _validate_steps' global id set), and gate both alias-write sites on
   it via a new alias_may_collide flag -- set only for fan-out's own
   template execution, so ordinary while/do-while loop-body aliasing
   (always globally unique by validation) is unaffected.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01U74yBbvVQCPwB7Ed8Dzeu6
---
 src/specify_cli/workflows/base.py   |   9 ++
 src/specify_cli/workflows/engine.py |  93 ++++++++++++++++-
 tests/test_workflows.py             | 148 ++++++++++++++++++++++++++++
 3 files changed, 249 insertions(+), 1 deletion(-)

diff --git a/src/specify_cli/workflows/base.py b/src/specify_cli/workflows/base.py
index 2466db8b1f..b650b26645 100644
--- a/src/specify_cli/workflows/base.py
+++ b/src/specify_cli/workflows/base.py
@@ -80,6 +80,15 @@ class StepContext:
     #: Source directory of the workflow definition file.
     workflow_dir: str | None = None
 
+    #: Every step id declared in the workflow OUTSIDE a fan-out template
+    #: (computed once per run from the workflow definition). Fan-out
+    #: templates are deliberately exempt from the global id-uniqueness
+    #: check, so a bare id inside one can collide with a real, distinct
+    #: step id in this set; the engine checks membership here before
+    #: writing a fan-out alias to a bare id, so that write can never
+    #: silently clobber an unrelated step's result.
+    reserved_step_ids: frozenset[str] = field(default_factory=frozenset)
+
 
 @dataclass
 class StepResult:
diff --git a/src/specify_cli/workflows/engine.py b/src/specify_cli/workflows/engine.py
index a50a676f3c..b79e1389cd 100644
--- a/src/specify_cli/workflows/engine.py
+++ b/src/specify_cli/workflows/engine.py
@@ -922,6 +922,10 @@ def _rename_step_tree_ids(
     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.
+
+    A nested ``while``/``do-while`` step's own id is still renamed, but its
+    ``steps`` body is deliberately left untouched — see the check below —
+    because that body gets its own runtime namespacing each iteration.
     """
     new_step = dict(step)
     id_map: dict[str, str] = {}
@@ -930,6 +934,26 @@ def _rename_step_tree_ids(
         new_id = f"{prefix}:{orig_id}:{suffix}"
         new_step["id"] = new_id
         id_map[new_id] = orig_id
+
+    # A while/do-while step re-namespaces its OWN 'steps' body per iteration
+    # at runtime (see the while/do-while branch in _execute_steps), each
+    # time treating whatever currently sits in each nested step's 'id' as
+    # the canonical original. Recursing into that body here too -- e.g.
+    # because this while step sits inside an outer loop iteration or
+    # fan-out item that is itself being namespaced right now -- would
+    # pre-namespace it once, so the loop's own per-iteration rename would
+    # then treat this already-namespaced id as "original" and alias back
+    # to *that* instead of the workflow author's real bare id: a 'leaf'
+    # step inside a while nested in a fan-out becomes 'fan:leaf:0' here,
+    # then 'fan:while:0:fan:leaf:0:0' there, aliased only back to
+    # 'fan:leaf:0' -- a synthetic id no ``steps.leaf`` reference resolves
+    # to. Rename this step's own id (above) so it still gets a unique id
+    # per outer iteration/item, but leave its body untouched so the loop's
+    # own runtime namespacing renames it against the real original ids
+    # exactly once.
+    if new_step.get("type") in ("while", "do-while"):
+        return new_step, id_map
+
     for key in _NESTED_STEP_LIST_KEYS:
         nested = new_step.get(key)
         if isinstance(nested, list):
@@ -962,6 +986,41 @@ def _rename_step_tree_ids(
     return new_step, id_map
 
 
+def _collect_reserved_step_ids(steps: list[dict[str, Any]]) -> frozenset[str]:
+    """Collect every step id declared outside a fan-out template.
+
+    Mirrors ``_validate_steps``'s global ``seen_ids``: walks the same
+    regular control-flow nesting keys (``then``/``else``/``steps``/
+    ``default``/``cases.*``), which validation requires to be globally
+    unique, but deliberately does NOT walk a fan-out's ``step`` template —
+    validation checks that subtree against a fresh, throwaway id set
+    specifically because the engine namespaces it at runtime (see
+    ``_run_fan_out``). The result is the set of ids a fan-out's bare-id
+    aliasing convenience write must never clobber: anything in this set is
+    a real, distinctly-authored step, not an artifact of the exempted
+    template.
+    """
+    ids: set[str] = set()
+    for step in steps:
+        if not isinstance(step, dict):
+            continue
+        step_id = step.get("id")
+        if isinstance(step_id, str):
+            ids.add(step_id)
+        for key in _NESTED_STEP_LIST_KEYS:
+            nested = step.get(key)
+            if isinstance(nested, list):
+                ids.update(_collect_reserved_step_ids(nested))
+        cases = step.get("cases")
+        if isinstance(cases, dict):
+            for case_steps in cases.values():
+                if isinstance(case_steps, list):
+                    ids.update(_collect_reserved_step_ids(case_steps))
+        # step.get("step") -- a fan-out's own template -- is deliberately
+        # not walked; see docstring.
+    return frozenset(ids)
+
+
 # -- Workflow Engine ------------------------------------------------------
 
 
@@ -1115,6 +1174,7 @@ def execute(
             project_root=str(self.project_root),
             run_id=state.run_id,
             workflow_dir=workflow_dir,
+            reserved_step_ids=_collect_reserved_step_ids(definition.steps),
         )
 
         # Execute steps
@@ -1186,6 +1246,7 @@ def resume(
             project_root=str(self.project_root),
             run_id=state.run_id,
             workflow_dir=state.workflow_dir,
+            reserved_step_ids=_collect_reserved_step_ids(definition.steps),
         )
 
         from . import STEP_REGISTRY
@@ -1248,6 +1309,7 @@ def _execute_steps(
         step_offset: int = 0,
         alias_map: dict[str, str] | None = None,
         alias_local_only: bool = False,
+        alias_may_collide: bool = False,
     ) -> None:
         """Execute a list of steps sequentially.
 
@@ -1266,6 +1328,20 @@ def _execute_steps(
         a private overlay as ``context.steps`` and this flag so concurrently
         running items never race to write the same original id in shared
         state; see ``_run_fan_out``.
+
+        ``alias_may_collide`` marks an ``alias_map`` whose original ids came
+        from inside a fan-out template — ids there are exempt from the
+        workflow's global id-uniqueness validation (see
+        ``_collect_reserved_step_ids``), so unlike a while/do-while loop
+        body's ids (always globally unique), one CAN collide with an
+        unrelated, distinctly-authored step's id. When set, the immediate
+        ``state.step_results`` mirror below is skipped for any original id
+        that is a member of ``context.reserved_step_ids`` — that id belongs
+        to a real step elsewhere in the workflow, and this alias write must
+        never clobber it. Propagated through the recursive calls below so it
+        stays set for descendants (e.g. a while loop nested inside the
+        fan-out template) once a fan-out template is entered; see
+        ``_run_fan_out``.
         """
         for i, step_config in enumerate(steps):
             step_id = step_config.get("id", f"step-{i}")
@@ -1326,7 +1402,9 @@ def _execute_steps(
                 if orig_id is not None:
                     if alias_local_only:
                         context.steps[orig_id] = step_data
-                    else:
+                    elif not (
+                        alias_may_collide and orig_id in context.reserved_step_ids
+                    ):
                         self._record_result(context, state, orig_id, step_data)
 
             state.append_log(
@@ -1460,6 +1538,7 @@ def _execute_steps(
                                 [ns_copy], context, state, registry,
                                 step_offset=-1, alias_map=id_map,
                                 alias_local_only=alias_local_only,
+                                alias_may_collide=alias_may_collide,
                             )
                             if state.status in (
                                 RunStatus.PAUSED,
@@ -1472,6 +1551,7 @@ def _execute_steps(
                         result.next_steps, context, state, registry,
                         step_offset=-1, alias_map=alias_map,
                         alias_local_only=alias_local_only,
+                        alias_may_collide=alias_may_collide,
                     )
                     if state.status in (
                         RunStatus.PAUSED,
@@ -1601,6 +1681,7 @@ def run_item(
                 self._execute_steps(
                     [item_step], item_ctx, state, registry, step_offset=-1,
                     alias_map=id_map, alias_local_only=local_only,
+                    alias_may_collide=True,
                 )
             finally:
                 item_ctx.steps = original_steps
@@ -1749,6 +1830,16 @@ def item_halt_status(idx: int) -> RunStatus | None:
         last_idx = halt[0] if halt is not None else (collected - 1 if collected else None)
         if last_idx is not None:
             for orig_id, data in alias_slots[last_idx].items():
+                # orig_id came from inside this fan-out's template, which is
+                # exempt from the global id-uniqueness check (see
+                # _collect_reserved_step_ids) -- it can coincide with a real,
+                # distinctly-authored step's id elsewhere in the workflow.
+                # The namespaced entry (written unconditionally above, via
+                # _record_result inside run_item) always survives regardless;
+                # skip only this bare-id convenience alias so it can never
+                # clobber that unrelated step's result.
+                if orig_id in context.reserved_step_ids:
+                    continue
                 self._record_result(context, state, orig_id, data)
 
         if halt is not None:
diff --git a/tests/test_workflows.py b/tests/test_workflows.py
index 8c9135419f..5d791bea8e 100644
--- a/tests/test_workflows.py
+++ b/tests/test_workflows.py
@@ -6489,6 +6489,98 @@ def execute(self, config, context):
         for i in items:
             assert state.step_results[f"fan:second:{i}"]["output"]["seen"] == i
 
+    def test_fan_out_concurrent_alias_never_clobbers_unrelated_step(self, tmp_path):
+        """A fan-out template's bare-id convenience alias must never
+        overwrite an unrelated, distinctly-authored step's result just
+        because the template's id happens to collide with it.
+
+        Fan-out template ids are exempt from the workflow's global
+        id-uniqueness validation (the engine's parentId:templateId:index key
+        is assumed to make collisions safe) -- so nothing stops a template
+        step from reusing an id already used by a real step elsewhere in the
+        workflow. This pre-populates `state.step_results["leaf"]` the way an
+        earlier, unrelated step would have, and marks "leaf" as reserved
+        (what `_collect_reserved_step_ids` would compute from the full
+        workflow definition), then runs a CONCURRENT fan-out whose template
+        also uses id "leaf". The unrelated entry must survive untouched;
+        only the namespaced per-item entries should exist for the fan-out's
+        own use of that id.
+        """
+        from specify_cli.workflows.base import RunStatus, StepBase, StepContext, StepResult, StepStatus
+        from specify_cli.workflows.engine import RunState, WorkflowEngine
+
+        class _LeafStep(StepBase):
+            type_key = "leaf-step"
+
+            def execute(self, config, context):
+                return StepResult(
+                    status=StepStatus.COMPLETED, output={"marker": context.item}
+                )
+
+        engine = WorkflowEngine(project_root=tmp_path)
+        unrelated_result = {
+            "type": "command",
+            "output": {"marker": "unrelated"},
+            "status": "completed",
+            "error": None,
+        }
+        context = StepContext(reserved_step_ids=frozenset({"leaf"}))
+        context.steps["leaf"] = unrelated_result
+        state = RunState(run_id="r", workflow_id="w", project_root=tmp_path)
+        state.status = RunStatus.RUNNING
+        state.step_results["leaf"] = unrelated_result
+        registry = {"leaf-step": _LeafStep()}
+        template = {"id": "leaf", "type": "leaf-step"}
+        items = ["a", "b", "c"]
+        engine._run_fan_out(items, template, "fan", context, state, registry, 3)
+
+        # The unrelated step's own result is untouched.
+        assert state.step_results["leaf"] == unrelated_result
+        assert context.steps["leaf"] == unrelated_result
+        # Each item's namespaced result still exists.
+        for idx, item in enumerate(items):
+            assert state.step_results[f"fan:leaf:{idx}"]["output"]["marker"] == item
+
+    def test_fan_out_sequential_alias_never_clobbers_unrelated_step(self, tmp_path):
+        """The same collision safety as
+        `test_fan_out_concurrent_alias_never_clobbers_unrelated_step`, but
+        for the SEQUENTIAL fan-out path (`max_concurrency` <= 1), which
+        writes its bare-id alias immediately per item rather than deferring
+        to a single post-join write.
+        """
+        from specify_cli.workflows.base import RunStatus, StepBase, StepContext, StepResult, StepStatus
+        from specify_cli.workflows.engine import RunState, WorkflowEngine
+
+        class _LeafStep(StepBase):
+            type_key = "leaf-step"
+
+            def execute(self, config, context):
+                return StepResult(
+                    status=StepStatus.COMPLETED, output={"marker": context.item}
+                )
+
+        engine = WorkflowEngine(project_root=tmp_path)
+        unrelated_result = {
+            "type": "command",
+            "output": {"marker": "unrelated"},
+            "status": "completed",
+            "error": None,
+        }
+        context = StepContext(reserved_step_ids=frozenset({"leaf"}))
+        context.steps["leaf"] = unrelated_result
+        state = RunState(run_id="r", workflow_id="w", project_root=tmp_path)
+        state.status = RunStatus.RUNNING
+        state.step_results["leaf"] = unrelated_result
+        registry = {"leaf-step": _LeafStep()}
+        template = {"id": "leaf", "type": "leaf-step"}
+        items = ["a", "b", "c"]
+        engine._run_fan_out(items, template, "fan", context, state, registry, 1)
+
+        assert state.step_results["leaf"] == unrelated_result
+        assert context.steps["leaf"] == unrelated_result
+        for idx, item in enumerate(items):
+            assert state.step_results[f"fan:leaf:{idx}"]["output"]["marker"] == item
+
     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
@@ -6537,6 +6629,62 @@ def test_fan_out_namespaces_nested_descendant_steps(self, project_dir):
         assert state.step_results["fan:leaf:1"]["output"]["stdout"] == "b\n"
         assert state.step_results["fan:leaf:2"]["output"]["stdout"] == "c\n"
 
+    def test_while_loop_nested_in_fan_out_aliases_to_true_original_id(
+        self, project_dir
+    ):
+        """A `while` loop that is itself a fan-out template (or nested inside
+        one) must alias its body's steps back to their real, bare original
+        ids -- not to an already-namespaced id produced by the fan-out's own
+        outer rename pass.
+
+        The fan-out's rename recurses into the template's subtree to
+        namespace steps nested arbitrarily deep (see
+        `test_fan_out_namespaces_nested_descendant_steps`). If that recursion
+        also renamed a nested `while` step's OWN `steps` body, the body's
+        `leaf` step would already be `fan:leaf:0` by the time the while
+        step's own per-iteration rename ran -- which would then treat
+        `fan:leaf:0` as "the original" and alias a doubly-prefixed id
+        (`fan:item:0:fan:leaf:0:0`) back to it instead of to the workflow
+        author's actual bare id `leaf`. `state.step_results["leaf"]` would
+        then never be populated at all, so `steps.leaf` could never resolve.
+        """
+        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-while"
+  name: "Fan Out Nested While"
+  version: "1.0.0"
+steps:
+  - id: fan
+    type: fan-out
+    items: "{{ ['a', 'b'] }}"
+    max_concurrency: 1
+    step:
+      id: item
+      type: while
+      condition: "true"
+      max_iterations: 1
+      steps:
+        - 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
+        # The bare id resolves -- to the latest (last) item's value, matching
+        # the alias convention used everywhere else in this module.
+        assert state.step_results["leaf"]["output"]["stdout"] == "b\n"
+        # Each item's own namespaced entry is still separately recoverable,
+        # aliased against its item-level prefix, not doubly-prefixed.
+        assert state.step_results["fan:item:0:leaf:0"]["output"]["stdout"] == "a\n"
+        assert state.step_results["fan:item:1:leaf:0"]["output"]["stdout"] == "b\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.

From a8188b60733933f2516d3b31bdc9cdbf064cff21 Mon Sep 17 00:00:00 2001
From: Noor-ul-ain001 <noor01mk@gmail.com>
Date: Sun, 13 Sep 2026 20:22:52 +0500
Subject: [PATCH 4/4] fix(workflows): stop ChainMap from breaking steps.*
 expressions in fan-out

The concurrent fan-out item isolation gave each item's context.steps a
ChainMap(local_overlay, original_steps) so a sibling step could resolve
an earlier one by its bare id without racing other concurrently-running
items. But _resolve_dot_path (the function every {{ steps.x.output... }}
expression goes through) only descends when isinstance(current, dict)
is true, and ChainMap is not a dict subclass -- so _build_namespace's
`ns["steps"] = context.steps or {}` put a non-dict at "steps", and every
steps.* expression evaluated inside a concurrent fan-out item silently
resolved to None. The existing test only exercised context.steps.get()
directly, which IS supported by ChainMap, so it never caught this.

Replaced the ChainMap with a plain dict snapshot of the shared steps
dict, taken once when the item starts. It still isolates this item's
own writes from concurrently-running siblings (a fresh copy per item,
never shared), it's a real dict so isinstance(..., dict) and every
expression path work normally, and the snapshot copy is safe under the
GIL against a concurrent sibling's writes to the source dict.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01U74yBbvVQCPwB7Ed8Dzeu6
---
 src/specify_cli/workflows/engine.py | 25 ++++++----
 tests/test_workflows.py             | 71 +++++++++++++++++++++++++++++
 2 files changed, 88 insertions(+), 8 deletions(-)

diff --git a/src/specify_cli/workflows/engine.py b/src/specify_cli/workflows/engine.py
index b79e1389cd..73cdc1948f 100644
--- a/src/specify_cli/workflows/engine.py
+++ b/src/specify_cli/workflows/engine.py
@@ -17,7 +17,6 @@
 import tempfile
 import threading
 import uuid
-from collections import ChainMap
 from concurrent.futures import Future, ThreadPoolExecutor
 from datetime import datetime, timezone
 from pathlib import Path
@@ -1660,23 +1659,33 @@ def run_item(
                 template, step_id, str(idx), default_id=base_id,
             )
             # ``local_only`` (concurrent path): give this item a private
-            # overlay for its ``.steps`` reads/writes. Namespaced results
+            # dict for its ``.steps`` reads/writes, snapshotting the shared
+            # steps dict at the point this item starts. Namespaced results
             # still land in the real ``state.step_results`` (via
             # _record_result's unconditional write — see _execute_steps),
             # but the immediate bare-id alias (see alias_map below) writes
-            # only into this overlay. That lets a later sibling step in
+            # only into this snapshot. That lets a later sibling step in
             # THIS item's template resolve an earlier sibling by its
-            # original id via the overlay, without ever mutating the
+            # original id via the snapshot, without ever mutating the
             # shared steps dict that other concurrently-running items also
             # read from — the actual race Copilot flagged: every worker
             # writing the same bare-id key could otherwise expose another
             # item's value to a sibling read. The caller applies exactly
             # one item's aliases to shared state — deterministically the
             # last item in item order — once every item has finished.
+            #
+            # A plain dict copy, not a ``ChainMap`` overlay: expression
+            # interpolation (``{{ steps.x.output... }}``, via
+            # ``_resolve_dot_path``) only descends through
+            # ``isinstance(current, dict)``, and ``ChainMap`` is not a
+            # ``dict`` subclass — every such expression evaluated inside a
+            # concurrent fan-out item would silently resolve to ``None``.
+            # The snapshot never sees an in-flight sibling item's writes to
+            # the shared dict made after this item started, but fan-out
+            # items were never entitled to see those anyway.
             original_steps = item_ctx.steps
-            local_overlay: dict[str, dict[str, Any]] = {}
-            if local_only:
-                item_ctx.steps = ChainMap(local_overlay, original_steps)
+            item_steps = dict(original_steps) if local_only else original_steps
+            item_ctx.steps = item_steps
             try:
                 self._execute_steps(
                     [item_step], item_ctx, state, registry, step_offset=-1,
@@ -1686,7 +1695,7 @@ def run_item(
             finally:
                 item_ctx.steps = original_steps
             alias_records: dict[str, dict[str, Any]] = {}
-            steps_view = local_overlay if local_only else item_ctx.steps
+            steps_view = item_steps if local_only else item_ctx.steps
             for new_id, orig_id in id_map.items():
                 if new_id in steps_view:
                     data = steps_view[new_id]
diff --git a/tests/test_workflows.py b/tests/test_workflows.py
index 5d791bea8e..a2917c607b 100644
--- a/tests/test_workflows.py
+++ b/tests/test_workflows.py
@@ -6489,6 +6489,77 @@ def execute(self, config, context):
         for i in items:
             assert state.step_results[f"fan:second:{i}"]["output"]["seen"] == i
 
+    def test_fan_out_concurrent_sibling_step_resolves_via_expression(self, tmp_path):
+        """A `{{ steps.<id>.output... }}` expression -- the real templating
+        path workflow YAML actually uses -- must resolve inside a
+        concurrent fan-out item, not just a direct `context.steps.get()`
+        Python call.
+
+        `_resolve_dot_path` (which every `{{ }}` expression goes through)
+        only descends through `isinstance(current, dict)`. The concurrent
+        item isolation previously gave each item a `ChainMap` overlay for
+        `context.steps` -- `ChainMap` is not a `dict` subclass, so
+        `_build_namespace`'s `ns["steps"] = context.steps or {}` put a
+        non-dict object at `steps`, and every `steps.*` expression
+        evaluated inside a concurrent fan-out item silently resolved to
+        `None`. A direct `.get()` call (as in
+        `test_fan_out_concurrent_sibling_step_isolated_per_item`) doesn't
+        exercise this, since `ChainMap` supports `.get()` directly.
+        """
+        import threading
+
+        from specify_cli.workflows.base import (
+            RunStatus,
+            StepBase,
+            StepContext,
+            StepResult,
+            StepStatus,
+        )
+        from specify_cli.workflows.engine import RunState, WorkflowEngine
+        from specify_cli.workflows.expressions import evaluate_expression
+        from specify_cli.workflows.steps.if_then import IfThenStep
+
+        n = 4
+        barrier = threading.Barrier(n, timeout=5)
+
+        class _WriteStep(StepBase):
+            type_key = "write"
+
+            def execute(self, config, context):
+                barrier.wait()
+                return StepResult(
+                    status=StepStatus.COMPLETED, output={"marker": context.item}
+                )
+
+        class _ReadStep(StepBase):
+            type_key = "read"
+
+            def execute(self, config, context):
+                seen = evaluate_expression(
+                    "{{ steps.first.output.marker }}", context
+                )
+                return StepResult(status=StepStatus.COMPLETED, output={"seen": seen})
+
+        engine = WorkflowEngine(project_root=tmp_path)
+        context = StepContext()
+        state = RunState(run_id="r", workflow_id="w", project_root=tmp_path)
+        state.status = RunStatus.RUNNING
+        registry = {"if": IfThenStep(), "write": _WriteStep(), "read": _ReadStep()}
+        template = {
+            "id": "item",
+            "type": "if",
+            "condition": "true",
+            "then": [
+                {"id": "first", "type": "write"},
+                {"id": "second", "type": "read"},
+            ],
+        }
+        items = list(range(n))
+        engine._run_fan_out(items, template, "fan", context, state, registry, n)
+
+        for i in items:
+            assert state.step_results[f"fan:second:{i}"]["output"]["seen"] == i
+
     def test_fan_out_concurrent_alias_never_clobbers_unrelated_step(self, tmp_path):
         """A fan-out template's bare-id convenience alias must never
         overwrite an unrelated, distinctly-authored step's result just