perf(workflows): copy a fixed pre-fan-out snapshot per concurrent item

Concurrent fan-out items each copied the live shared steps dict, which
completed items keep growing with their namespaced results. With a small
worker count that made every later item's copy larger, O(K * N^2) work
overall, and what a late-starting item saw depended on scheduling. The
concurrent path now takes one snapshot before any worker starts and each
item copies that.

Also corrects the run_item comment that still described applying only
the last item's aliases (the post-join code folds every collected item's
aliases in item order), and adds an end-to-end resume() test for the
reserved-id collision guard, since resume() builds its own run context
separately from execute().

Assisted-by: Claude Code (model: Claude Opus 5.5, autonomous)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: Noor-ul-ain001

src/specify_cli/workflows/engine.py

@@ -1791,7 +1791,11 @@ def item_id(idx: int) -> str:
             return f"{step_id}:{base_id}:{idx}"
 
         def run_item(
-            idx: int, item_ctx: StepContext, *, local_only: bool
+            idx: int,
+            item_ctx: StepContext,
+            *,
+            local_only: bool,
+            base_steps: dict[str, Any] | None = None,
         ) -> 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
@@ -1820,9 +1824,21 @@ def run_item(
             # 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.
+            # item's value to a sibling read. Each item returns its aliases
+            # instead; the concurrent caller folds every collected item's
+            # aliases in item order once the pool has joined (so an id
+            # resolves to the latest item that actually wrote it), and the
+            # sequential caller publishes them after each item.
+            #
+            # ``base_steps``, when given, is the dict to snapshot instead of
+            # the live ``original_steps``. The concurrent path passes one
+            # snapshot taken before any item starts: completed items keep
+            # publishing their namespaced keys into ``original_steps``
+            # (below), so copying it per item would cost O(items x keys)
+            # per copy and O(K * N^2) overall, and what a late-starting item
+            # saw would depend on scheduling. Copying the fixed pre-fan-out
+            # snapshot keeps each copy the same size and makes every
+            # concurrent item see the same namespace.
             #
             # A plain dict copy, not a ``ChainMap`` overlay: expression
             # interpolation (``{{ steps.x.output... }}``, via
@@ -1834,7 +1850,12 @@ def run_item(
             # the shared dict made after this item started, but fan-out
             # items were never entitled to see those anyway.
             original_steps = item_ctx.steps
-            item_steps = dict(original_steps) if local_only else original_steps
+            if local_only:
+                item_steps = dict(
+                    original_steps if base_steps is None else base_steps
+                )
+            else:
+                item_steps = original_steps
             item_ctx.steps = item_steps
 
             # Sequential items (not local_only) execute directly against the
@@ -1954,6 +1975,10 @@ def run_item(
         n = len(items)
         slots: list[Any] = [None] * n
         alias_slots: list[dict[str, dict[str, Any]]] = [{}] * n
+        # Taken once, before any worker starts, so every item copies the
+        # same fixed-size namespace rather than the live dict that completed
+        # items keep growing (see ``base_steps`` in run_item).
+        base_steps = dict(context.steps)
 
         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
@@ -1968,6 +1993,7 @@ def run_isolated(idx: int) -> tuple[Any, dict[str, dict[str, Any]]]:
                     inside_fan_out=True,
                 ),
                 local_only=True,
+                base_steps=base_steps,
             )
 
         def item_halt_status(idx: int) -> RunStatus | None:

tests/test_workflows.py

@@ -7385,6 +7385,112 @@ def test_fan_out_reserved_id_collision_end_to_end(
             )
         assert state.step_results["after"]["output"]["stdout"].strip() == "outside"
 
+    @pytest.mark.parametrize("max_concurrency", [1, 2])
+    def test_fan_out_reserved_id_collision_end_to_end_on_resume(
+        self, project_dir, max_concurrency
+    ):
+        """`resume()` builds its own run context, separately from
+        `execute()`, so it needs its own end-to-end coverage of the
+        `reserved_step_ids` wiring.
+
+        The outside `leaf` step completes, a gate pauses the run, and the
+        colliding fan-out only runs after `engine.resume()`. On resume
+        `context.steps` IS `state.step_results`, so a missing reserved set
+        there would let the fan-out's bare `leaf` alias overwrite the
+        persisted outside result directly.
+        """
+        from unittest.mock import patch
+
+        from specify_cli.workflows.base import RunStatus, StepResult
+        from specify_cli.workflows.engine import WorkflowDefinition, WorkflowEngine
+
+        yaml_str = f"""
+schema_version: "1.0"
+workflow:
+  id: "fan-out-reserved-collision-resume"
+  name: "Fan Out Reserved Collision Resume"
+  version: "1.0.0"
+steps:
+  - id: leaf
+    type: shell
+    run: "echo outside"
+  - id: approve
+    type: gate
+    message: "Approve?"
+  - id: fan
+    type: fan-out
+    items: "{{{{ ['a', 'b', 'c'] }}}}"
+    max_concurrency: {max_concurrency}
+    step:
+      id: leaf
+      type: shell
+      run: "echo {{{{ item }}}}"
+  - id: after
+    type: shell
+    run: "echo {{{{ steps.leaf.output.stdout }}}}"
+"""
+        definition = WorkflowDefinition.from_string(yaml_str)
+        engine = WorkflowEngine(project_dir)
+        state = engine.execute(definition)
+        assert state.status == RunStatus.PAUSED
+
+        with patch(
+            "specify_cli.workflows.step.gate.GateStep.execute",
+            return_value=StepResult(output={"approved": True}),
+        ):
+            state = engine.resume(state.run_id)
+
+        assert state.status == RunStatus.COMPLETED
+        assert state.step_results["leaf"]["output"]["stdout"] == "outside\n"
+        for idx, item in enumerate(["a", "b", "c"]):
+            assert (
+                state.step_results[f"fan:leaf:{idx}"]["output"]["stdout"]
+                == f"{item}\n"
+            )
+        assert state.step_results["after"]["output"]["stdout"].strip() == "outside"
+
+    def test_concurrent_fan_out_items_copy_fixed_pre_fan_out_snapshot(
+        self, tmp_path
+    ):
+        """Every concurrent item must copy the namespace as it was before
+        the fan-out started, not the live steps dict.
+
+        Completed items publish their namespaced keys back into the shared
+        steps dict. With 2 workers, item 2 is only submitted once item 0
+        has been collected, so if items copied the live dict, item 2 would
+        see `fan:probe:0`, item 4 would see three earlier keys, and so on.
+        That makes the copies grow with the item count (quadratic overall)
+        and makes what an item sees depend on scheduling.
+        """
+        from specify_cli.workflows.base import RunStatus, StepBase, StepContext, StepResult, StepStatus
+        from specify_cli.workflows.engine import RunState, WorkflowEngine
+
+        seen: dict[int, list[str]] = {}
+
+        class _ProbeStep(StepBase):
+            type_key = "probe"
+
+            def execute(self, config, context):
+                seen[context.item] = sorted(
+                    k for k in context.steps if k.startswith("fan:")
+                )
+                return StepResult(status=StepStatus.COMPLETED, output={})
+
+        engine = WorkflowEngine(project_root=tmp_path)
+        context = StepContext()
+        context.steps["before"] = {"output": {}}
+        state = RunState(run_id="r", workflow_id="w", project_root=tmp_path)
+        state.status = RunStatus.RUNNING
+        registry = {"probe": _ProbeStep()}
+        template = {"id": "probe", "type": "probe"}
+        items = list(range(6))
+        engine._run_fan_out(items, template, "fan", context, state, registry, 2)
+
+        assert seen == {i: [] for i in items}
+        # Every item's namespaced result is still published afterwards.
+        for i in items:
+            assert f"fan:probe:{i}" in context.steps
+
     @pytest.mark.parametrize("inner_concurrency", [1, 2])
     def test_reserved_alias_visible_after_nested_fan_out_in_sequential_item(
         self, project_dir, inner_concurrency