storing seed consistently

Author: networmix

ngraph/results/store.py

@@ -18,11 +18,24 @@ class WorkflowStepMetadata:
         step_type: The workflow step class name (e.g., 'CapacityEnvelopeAnalysis').
         step_name: The instance name of the step.
         execution_order: Order in which this step was executed (0-based).
+        scenario_seed: Scenario-level seed provided in the YAML (if any).
+        step_seed: Seed assigned to this step (explicit or scenario-derived).
+        seed_source: Source for the step seed. One of:
+            - "scenario-derived": seed was derived from scenario.seed
+            - "explicit-step": seed was explicitly provided for the step
+            - "none": no seed provided/active for this step
+        active_seed: The effective base seed used by the step, if any. For steps
+            that use Monte Carlo execution, per-iteration seeds are derived from
+            active_seed (e.g., active_seed + iteration_index).
     """
 
     step_type: str
     step_name: str
     execution_order: int
+    scenario_seed: Optional[int] = None
+    step_seed: Optional[int] = None
+    seed_source: str = "none"
+    active_seed: Optional[int] = None
 
 
 @dataclass
@@ -61,17 +74,35 @@ def put(self, step_name: str, key: str, value: Any) -> None:
         self._store[step_name][key] = value
 
     def put_step_metadata(
-        self, step_name: str, step_type: str, execution_order: int
+        self,
+        step_name: str,
+        step_type: str,
+        execution_order: int,
+        *,
+        scenario_seed: Optional[int] = None,
+        step_seed: Optional[int] = None,
+        seed_source: str = "none",
+        active_seed: Optional[int] = None,
     ) -> None:
         """Store metadata for a workflow step.
 
         Args:
             step_name: The step instance name.
             step_type: The workflow step class name.
             execution_order: Order in which this step was executed (0-based).
+            scenario_seed: Scenario-level seed from YAML, if any.
+            step_seed: Seed attached to this step (explicit or derived), if any.
+            seed_source: Source of step seed ("scenario-derived", "explicit-step", or "none").
+            active_seed: Effective base seed used by the step, if any.
         """
         self._metadata[step_name] = WorkflowStepMetadata(
-            step_type=step_type, step_name=step_name, execution_order=execution_order
+            step_type=step_type,
+            step_name=step_name,
+            execution_order=execution_order,
+            scenario_seed=scenario_seed,
+            step_seed=step_seed,
+            seed_source=seed_source,
+            active_seed=active_seed,
         )
 
     def get(self, step_name: str, key: str, default: Any = None) -> Any:
@@ -148,6 +179,10 @@ def to_dict(self) -> Dict[str, Any]:
                 "step_type": metadata.step_type,
                 "step_name": metadata.step_name,
                 "execution_order": metadata.execution_order,
+                "scenario_seed": metadata.scenario_seed,
+                "step_seed": metadata.step_seed,
+                "seed_source": metadata.seed_source,
+                "active_seed": metadata.active_seed,
             }
             for step_name, metadata in self._metadata.items()
         }

ngraph/scenario.py

@@ -486,13 +486,25 @@ def _build_workflow_steps(
             # Ensure the constructed WorkflowStep receives the resolved unique name
             normalized_ctor_args["name"] = step_name
 
-            # Add seed derivation for workflow steps that don't have explicit seed
-            if "seed" not in normalized_ctor_args:
+            # Determine seed provenance and possibly derive a step seed
+            seed_source: str = "none"
+            if (
+                "seed" in normalized_ctor_args
+                and normalized_ctor_args["seed"] is not None
+            ):
+                seed_source = "explicit-step"
+            else:
                 derived_seed = seed_manager.derive_seed("workflow_step", step_name)
                 if derived_seed is not None:
                     normalized_ctor_args["seed"] = derived_seed
+                    seed_source = "scenario-derived"
 
             step_obj = step_cls(**normalized_ctor_args)
+            # Attach internal provenance for metadata collection
+            try:
+                step_obj._seed_source = seed_source
+            except Exception:
+                pass
             steps.append(step_obj)
 
         return steps

ngraph/workflow/base.py

@@ -70,6 +70,8 @@ class WorkflowStep(ABC):
 
     name: str = ""
     seed: Optional[int] = None
+    # Internal: provenance of the step seed ("explicit-step" or "scenario-derived" or "none").
+    _seed_source: str = ""
 
     def execute(self, scenario: "Scenario") -> None:
         """Execute the workflow step with logging and metadata storage.
@@ -94,9 +96,36 @@ def execute(self, scenario: "Scenario") -> None:
         step_name = self.name
         display_name = step_name or step_type
 
+        # Determine seed provenance and effective seed for this step
+        scenario_seed = getattr(scenario, "seed", None)
+        step_seed = self.seed
+        explicit_source = getattr(self, "_seed_source", None)
+        if step_seed is not None and explicit_source == "explicit-step":
+            seed_source = "explicit-step"
+            active_seed = step_seed
+        elif step_seed is not None and explicit_source == "scenario-derived":
+            # Step received a derived seed at construction time
+            seed_source = "scenario-derived"
+            active_seed = step_seed
+        elif scenario_seed is not None:
+            seed_source = "scenario-derived"
+            # Scenario.from_yaml derives per-step seeds when seed is provided; if a
+            # concrete seed was not set on the step (self.seed is None), treat the
+            # scenario seed as the active base (workers may derive offsets internally).
+            active_seed = scenario_seed
+        else:
+            seed_source = "none"
+            active_seed = None
+
         # Store workflow metadata before execution using the exact step namespace key
         scenario.results.put_step_metadata(
-            step_name=step_name, step_type=step_type, execution_order=_execution_counter
+            step_name=step_name,
+            step_type=step_type,
+            execution_order=_execution_counter,
+            scenario_seed=scenario_seed,
+            step_seed=step_seed,
+            seed_source=seed_source,
+            active_seed=active_seed,
         )
         _execution_counter += 1
 

tests/workflow/test_analysis_integration.py

@@ -145,6 +145,10 @@ def test_workflow_step_metadata_storage(self, simple_scenario):
         metadata = simple_scenario.results.get_step_metadata("meta_test")
         assert metadata is not None
         assert metadata.step_type == "NetworkStats"
+        # Seed metadata presence (scenario has seed set in helper scenarios)
+        assert hasattr(metadata, "scenario_seed")
+        assert hasattr(metadata, "step_seed")
+        assert hasattr(metadata, "seed_source")
         assert metadata.step_name == "meta_test"
         assert metadata.execution_order >= 0
 

tests/workflow/test_base.py

@@ -2,6 +2,7 @@
 
 import pytest
 
+from ngraph.scenario import Scenario
 from ngraph.workflow.base import (
     WORKFLOW_STEP_REGISTRY,
     WorkflowStep,
@@ -41,14 +42,38 @@ def test_workflow_step_subclass_run_method() -> None:
 
     class ConcreteStep(WorkflowStep):
         def run(self, scenario) -> None:
-            scenario.called = True
+            # Set a flag on the step instance to confirm invocation
+            self._ran = True
 
-    mock_scenario = MagicMock()
+    mock_scenario = MagicMock(spec=Scenario)
     step_instance = ConcreteStep(name="test_step")
     step_instance.run(mock_scenario)
 
     # Check if run() was actually invoked
-    # e.g., we set scenario.called = True in run()
-    # but here we can also rely on MagicMock calls or attributes if needed
-    assert hasattr(mock_scenario, "called") and mock_scenario.called is True
+    assert getattr(step_instance, "_ran", False) is True
     assert step_instance.name == "test_step"
+
+
+def test_execute_records_metadata_including_seed_fields() -> None:
+    """Execute a minimal step and verify metadata includes seed fields."""
+    from ngraph.results import Results
+
+    class Dummy(WorkflowStep):
+        def run(self, scenario) -> None:
+            scenario.results.put(self.name, "ok", True)
+
+    scen = MagicMock(spec=Scenario)
+    scen.results = Results()
+    scen.seed = 1010
+    step = Dummy(name="d1")
+    step.execute(scen)
+
+    md = scen.results.get_step_metadata("d1")
+    assert md is not None
+    assert md.step_type == "Dummy"
+    assert md.step_name == "d1"
+    assert isinstance(md.execution_order, int) and md.execution_order >= 0
+    # New fields
+    assert hasattr(md, "scenario_seed") and md.scenario_seed == 1010
+    assert hasattr(md, "step_seed")
+    assert hasattr(md, "seed_source")

tests/workflow/test_capacity_envelope_analysis.py

@@ -189,6 +189,8 @@ def test_run_with_mock_failure_manager(
 
         # Verify results were processed (just check that the step ran without error)
         # The analysis and results storage happened as evidenced by the log messages
+        # Metadata is recorded by the execute() wrapper; here we called run() directly,
+        # so metadata may be absent. This check is only applicable when using execute().
 
     @patch("ngraph.workflow.capacity_envelope_analysis.FailureManager")
     def test_run_with_failure_patterns(self, mock_failure_manager_class, mock_scenario):

tests/workflow/test_msd_perf_safety.py

@@ -45,6 +45,10 @@ class _MD:
                 def __init__(self, execution_order: int, step_type: str) -> None:
                     self.execution_order = execution_order
                     self.step_type = step_type
+                    # Optional fields added in metadata; harmless defaults for stub
+                    self.scenario_seed = None
+                    self.step_seed = None
+                    self.seed_source = "none"
 
             return {"msd": _MD(0, "MaximumSupportedDemandAnalysis")}
 

tests/workflow/test_tm_analysis_perf_safety.py

@@ -46,6 +46,7 @@ def get(self, step: str, key: str) -> Any:
             return self._data.get(step, {}).get(key)
 
         def get_all_step_metadata(self):
+            # Return empty mapping; caller code should handle gracefully
             return {}
 
     class _FailurePolicySetStub: