every IndexedStep can now store its own variables

CHANGELOG.md

@@ -7,6 +7,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- Indexed steps can now store values. `local` and `global` output mappings on a
+  step with an `indexed: true` input are no longer discarded: the wrapper saves
+  the aggregated list of per-iteration values (one entry per iteration) under the
+  configured `local_name` / `global_name`. `passfail`, `equals`, `range` and
+  `image` outputs continue to be evaluated and attached per iteration.
+- An indexed step whose indexed lists are empty now stores empty lists instead of
+  erroring, and an iteration that errors no longer masks its own failure with a
+  `KeyError` while the aggregate is stored.
+
+### Removed
+
+- Handling of `indexed: true` on an *output* mapping in `IndexedStep`. Every
+  output model is `extra="forbid"`, so the field was rejected by validation and
+  the branch was unreachable from any valid recipe.
+
 ## [0.6.1] - 2026-08-27
 
 - Add CHANGELOG.md

docs/source/yaml_format.rst

@@ -72,6 +72,15 @@ verdict mapping on a step. ``local`` and ``global`` store values, while
 :ref:`recipe-v2-output-passfail` through :ref:`recipe-v2-output-range`
 definitions for exact fields.
 
+On a step with an indexed input, output mappings are split between the iterations
+and the wrapper. ``passfail``, ``equals``, ``range`` and ``image`` are evaluated
+(or attached to the report) once per iteration, so each iteration gets its own
+verdict row and its own images. ``local`` and ``global`` are stored once, by the
+wrapper, and receive the **list** of per-iteration values — one entry per
+iteration, in iteration order. An indexed step whose indexed lists are empty
+skips and stores empty lists. ``indexed`` is an input-only field; it is rejected
+on an output mapping.
+
 Runtime-specific behavior
 -------------------------
 

src/pypts/steps.py

@@ -20,6 +20,12 @@
 
 logger = logging.getLogger(__name__)
 
+# Output types evaluated per iteration by each generated sub-step.
+_PER_ITERATION_RESULT_TYPES = ("passthrough", "passfail", "equals", "range")
+# Output types the IndexedStep wrapper owns: values are aggregated across all
+# iterations and stored once, so the iterations must not write them N times.
+_STORAGE_OUTPUT_TYPES = ("local", "global")
+
 
 class IndexedStep(Step):
     """
@@ -40,11 +46,20 @@ def __init__(self, step: Step, **kwargs):
             raise TypeError(f"IndexedStep requires a valid Step instance, got {type(step)}")
         self.template_step: Step = step
         self.steps: List[Step] = [] # Stores the generated step instances for each run
-        # Override output_mapping for the wrapper step itself to capture the aggregate result
-        self.output_mapping = {"__result": {"type": "passthrough"}}
-        # Note: Any output mappings defined in the original step_data for the wrapper
-        # (like saving the aggregated result to a variable) should be added here or
-        # processed by the base Step.process_outputs method using self.output_mapping.
+        # The wrapper owns the storage (local/global) mappings: their values are
+        # aggregated over every iteration and saved once, from process_outputs.
+        # Verdict and image mappings stay on the template step so they are evaluated
+        # (or attached to the report) per iteration.
+        # A new dict is built on purpose: build_step hands the *same* output_mapping
+        # object to both the template step and this wrapper, so mutating in place would
+        # corrupt the template and break the aggregation below.
+        self.output_mapping = {
+            name: config
+            for name, config in self.output_mapping.items()
+            if isinstance(config, dict) and config.get("type") in _STORAGE_OUTPUT_TYPES
+        }
+        # The wrapper's own verdict is the aggregate of every iteration's result.
+        self.output_mapping["__result"] = {"type": "passthrough"}
 
     def check_indexing(self):
         """
@@ -79,6 +94,17 @@ def _step(self, runtime: Runtime, input: dict, parent_step_result_uuid: uuid.UUI
 
         indexed_list_names = list(indexed_input_configs.keys())
 
+        # Pre-seed the aggregation with an empty list per non-verdict output of the
+        # template step. Every key the wrapper's output_mapping may reference then
+        # always exists in the returned dict, even when an iteration errors (which
+        # leaves StepResult.outputs empty) or when there is nothing to run at all.
+        aggregated_outputs = {
+            name: []
+            for name, config in self.template_step.output_mapping.items()
+            if isinstance(config, dict)
+            and config.get("type") not in _PER_ITERATION_RESULT_TYPES
+        }
+
         # Get the actual input *values* provided to the IndexedStep wrapper
         wrapper_inputs = input
 
@@ -98,7 +124,9 @@ def _step(self, runtime: Runtime, input: dict, parent_step_result_uuid: uuid.UUI
                 num_runs = min(len(wrapper_inputs[name]) for name in indexed_list_names)
                 if num_runs == 0:
                     logger.warning(f"IndexedStep '{self.name}' has empty lists for indexed inputs. Skipping execution.")
-                    return {"__result": ResultType.SKIP} # Or DONE? SKIP seems appropriate.
+                    # Storage outputs still resolve — to empty lists — so the wrapper's
+                    # process_outputs does not fail on a missing key.
+                    return {**aggregated_outputs, "__result": ResultType.SKIP} # Or DONE? SKIP seems appropriate.
             except ValueError: # Handle case where indexed_list_names is empty after checks
                 logger.warning(f"IndexedStep '{self.name}' inconsistency: Indexed inputs found, but failed to determine run count. Running once.")
                 num_runs = 1
@@ -132,22 +160,15 @@ def _step(self, runtime: Runtime, input: dict, parent_step_result_uuid: uuid.UUI
             copied_step.input_mapping = iteration_input_mapping
 
             # Remove local/global variable *saving* definitions from the copied step's *output* mapping.
-            # We want to aggregate these values in the wrapper, not have each iteration save potentially overwriting variables.
-            # Pass/Fail/Range/Equals checks on outputs should still happen per iteration.
-            output_mapping_keys = list(copied_step.output_mapping.keys())
-            for key in output_mapping_keys:
-                output_conf = copied_step.output_mapping[key]
-                if isinstance(output_conf, dict):
-                    if output_conf.get("indexed", False):
-                        values = output_conf.get("value", [])
-                        if i < len(values):
-                            copied_step.output_mapping[key]["value"] = values[i]
-                        else:
-                            logger.warning(f"No indexed output value for iteration {i} in '{self.name}'")
-                            copied_step.output_mapping[key]["value"] = None #could also be a specified standard value.
-                    elif output_conf.get("type") in ["local", "global"]:
-                        logger.debug(f"Removing output mapping '{key}' (type: {output_conf.get('type')}) from iteration {i} of '{self.name}'")
-                        del copied_step.output_mapping[key]
+            # The wrapper aggregates these values and saves them once, instead of each
+            # iteration overwriting the same variable.
+            # Pass/Fail/Range/Equals checks and image outputs still happen per iteration.
+            copied_step.output_mapping = {
+                name: config
+                for name, config in copied_step.output_mapping.items()
+                if not (isinstance(config, dict)
+                        and config.get("type") in _STORAGE_OUTPUT_TYPES)
+            }
 
             # Modify the name for clarity in logs and results.
             # If step_name contains {input_name} placeholders, substitute
@@ -174,20 +195,12 @@ def _step(self, runtime: Runtime, input: dict, parent_step_result_uuid: uuid.UUI
         # Aggregate outputs from the individual step results.
         # We only aggregate outputs that were *not* used for pass/fail/range/equals checks
         # within the iterations (i.e., likely intended as data outputs).
-        aggregated_outputs = {}
-        # These types imply the output was used for per-iteration result calculation.
-        per_iteration_result_types = ["passthrough", "passfail", "equals", "range"]
-
+        # aggregated_outputs was pre-seeded above with the template's non-verdict
+        # outputs, so only those names are collected here.
         for result in step_results:
             # Iterate through the outputs recorded in the StepResult for this iteration
             for output_name, output_value in result.outputs.items():
-                # Check the *template* step's original output mapping config for this output name.
-                template_output_conf = self.template_step.output_mapping.get(output_name)
-
-                # If the output exists in the template config AND its type was NOT a per-iteration check, aggregate it.
-                if template_output_conf and isinstance(template_output_conf, dict) and template_output_conf.get("type") not in per_iteration_result_types:
-                    if output_name not in aggregated_outputs:
-                        aggregated_outputs[output_name] = []
+                if output_name in aggregated_outputs:
                     aggregated_outputs[output_name].append(output_value)
 
         # Determine the overall result of the IndexedStep based on all iteration results.

tests/unit_tests/test_steps.py

@@ -51,6 +51,20 @@ def _step(self, runtime, input, parent_step):
         return input
 
 
+class EchoWithVerdictStep(Step):
+    """Echoes the indexed input 'a' as a data output plus a passing verdict."""
+    def _step(self, runtime, input, parent_step):
+        return {"value": input["a"], "passed": True}
+
+
+class FlakyEchoStep(Step):
+    """Echoes the indexed input 'a', but raises when it is 20."""
+    def _step(self, runtime, input, parent_step):
+        if input["a"] == 20:
+            raise ValueError("deliberate failure")
+        return {"value": input["a"]}
+
+
 @pytest.fixture
 def mock_runtime():
     """A lightweight mock Runtime that satisfies Step.run's contract."""
@@ -423,6 +437,34 @@ def test_build_indexed_step_wraps_when_indexed(self):
         assert isinstance(step, IndexedStep)
         assert isinstance(step.template_step, WaitStep)
 
+    def test_build_indexed_step_keeps_storage_output_mapping(self):
+        """Storage outputs survive on the wrapper; verdicts stay on the template."""
+        from pypts.recipe_language import WaitStep as WaitDefinition
+
+        definition = WaitDefinition(
+            steptype="WaitStep",
+            step_name="IndexedWait",
+            description="Wait twice and store the waits.",
+            input_mapping={
+                "wait_time": {
+                    "type": "direct", "value": [0.01, 0.02], "indexed": True
+                }
+            },
+            output_mapping={
+                "waited": {"type": "local", "local_name": "waited"},
+                "passed": {"type": "passfail"},
+            },
+        )
+        step = Step.build_step(definition)
+        assert isinstance(step, IndexedStep)
+        assert step.output_mapping["waited"] == {
+            "type": "local", "local_name": "waited"
+        }
+        assert step.output_mapping["__result"] == {"type": "passthrough"}
+        assert "passed" not in step.output_mapping
+        # The template keeps the full authored mapping (shared-dict aliasing guard).
+        assert set(step.template_step.output_mapping) == {"waited", "passed"}
+
     def test_rejects_unvalidated_dictionary(self):
         with pytest.raises(TypeError, match="validated step definition"):
             Step.build_step({"steptype": "WaitStep"})
@@ -656,6 +698,164 @@ def test_empty_indexed_list_skips(self, mock_runtime):
         result = indexed.run(mock_runtime, {})
         assert result.result == ResultType.SKIP
 
+    # --- Output storage -------------------------------------------------
+
+    @staticmethod
+    def _prepare(runtime):
+        """Give a real Runtime the scopes and metadata Step.run needs."""
+        runtime.stop_event = Event()
+        runtime.push_locals({})
+        runtime.set_globals({})
+        runtime.recipe_name = "T"
+        runtime.recipe_file_name = "t.yaml"
+        runtime.serial_number = "SN"
+        runtime.current_sequence_name = "Main"
+
+    def test_local_output_stores_aggregated_list(self, real_runtime):
+        self._prepare(real_runtime)
+        # build_step passes one and the same mapping object to both, so do that here.
+        mapping = {"a": {"type": "local", "local_name": "measured"}}
+        template = PassFailStep(
+            step_name="Inner",
+            input_mapping={"a": {"type": "direct", "indexed": True}},
+            output_mapping=mapping,
+        )
+        indexed = IndexedStep(
+            template,
+            step_name="Outer",
+            input_mapping={
+                "a": {"type": "direct", "value": [10, 20, 30], "indexed": True}
+            },
+            output_mapping=mapping,
+        )
+
+        result = indexed.run(real_runtime, {})
+        assert real_runtime.get_local("measured") == [10, 20, 30]
+        assert result.outputs["a"] == [10, 20, 30]
+
+    def test_global_output_stores_aggregated_list(self, real_runtime):
+        self._prepare(real_runtime)
+        mapping = {"a": {"type": "global", "global_name": "saved"}}
+        template = PassFailStep(
+            step_name="Inner",
+            input_mapping={"a": {"type": "direct", "indexed": True}},
+            output_mapping=mapping,
+        )
+        indexed = IndexedStep(
+            template,
+            step_name="Outer",
+            input_mapping={
+                "a": {"type": "direct", "value": [1, 2], "indexed": True}
+            },
+            output_mapping=mapping,
+        )
+
+        indexed.run(real_runtime, {})
+        assert real_runtime.get_global("saved") == [1, 2]
+
+    def test_verdict_and_storage_coexist(self, real_runtime):
+        """A verdict output next to a storage output must not trip the
+        passthrough-exclusivity rule, and both must still take effect."""
+        self._prepare(real_runtime)
+        mapping = {
+            "passed": {"type": "passfail"},
+            "value": {"type": "local", "local_name": "measured"},
+        }
+        template = EchoWithVerdictStep(
+            step_name="Inner",
+            input_mapping={"a": {"type": "direct", "indexed": True}},
+            output_mapping=mapping,
+        )
+        indexed = IndexedStep(
+            template,
+            step_name="Outer",
+            input_mapping={
+                "a": {"type": "direct", "value": [10, 20], "indexed": True}
+            },
+            output_mapping=mapping,
+        )
+
+        result = indexed.run(real_runtime, {})
+        assert result.error_info == ""
+        assert result.result == ResultType.PASS
+        assert real_runtime.get_local("measured") == [10, 20]
+        # The verdict was evaluated once per iteration, on the sub-steps.
+        assert [sub.result for sub in result.subresults] == [
+            ResultType.PASS, ResultType.PASS
+        ]
+
+    def test_wrapper_and_template_mappings_are_split(self, mock_runtime):
+        mapping = {
+            "passed": {"type": "passfail"},
+            "value": {"type": "local", "local_name": "measured"},
+            "chart": {"type": "image"},
+        }
+        template = EchoWithVerdictStep(
+            step_name="Inner",
+            input_mapping={"a": {"type": "direct", "indexed": True}},
+            output_mapping=mapping,
+        )
+        indexed = IndexedStep(
+            template,
+            step_name="Outer",
+            input_mapping={
+                "a": {"type": "direct", "value": [10, 20], "indexed": True}
+            },
+            output_mapping=mapping,
+        )
+
+        # Wrapper keeps only storage mappings, plus its own aggregate verdict.
+        assert set(indexed.output_mapping) == {"value", "__result"}
+        # The shared authored mapping is left untouched on the template.
+        assert set(template.output_mapping) == {"passed", "value", "chart"}
+
+        indexed.run(mock_runtime, {})
+        for sub in indexed.steps:
+            assert set(sub.output_mapping) == {"passed", "chart"}
+
+    def test_empty_indexed_list_stores_empty_list(self, real_runtime):
+        self._prepare(real_runtime)
+        mapping = {"a": {"type": "local", "local_name": "measured"}}
+        template = PassFailStep(
+            step_name="Inner",
+            input_mapping={"a": {"type": "direct", "indexed": True}},
+            output_mapping=mapping,
+        )
+        indexed = IndexedStep(
+            template,
+            step_name="Outer",
+            input_mapping={"a": {"type": "direct", "value": [], "indexed": True}},
+            output_mapping=mapping,
+        )
+
+        result = indexed.run(real_runtime, {})
+        assert result.result == ResultType.SKIP
+        assert real_runtime.get_local("measured") == []
+
+    def test_errored_iteration_still_stores_list(self, real_runtime):
+        """A failing iteration must surface as ERROR, not be masked by a
+        KeyError while storing the aggregate."""
+        self._prepare(real_runtime)
+        mapping = {"value": {"type": "local", "local_name": "measured"}}
+        template = FlakyEchoStep(
+            step_name="Inner",
+            input_mapping={"a": {"type": "direct", "indexed": True}},
+            output_mapping=mapping,
+        )
+        indexed = IndexedStep(
+            template,
+            step_name="Outer",
+            input_mapping={
+                "a": {"type": "direct", "value": [10, 20, 30], "indexed": True}
+            },
+            output_mapping=mapping,
+        )
+
+        result = indexed.run(real_runtime, {})
+        assert result.result == ResultType.ERROR
+        assert result.error_info == ""  # the wrapper itself did not raise
+        assert real_runtime.get_local("measured") == [10]
+
 
 # ============================================================
 # PythonModuleStep