BUG: fix Monte Carlo seeding race and non-reproducible SeedSequence

Addresses review feedback on RocketPy-Team#1054.

Parallel workers claimed the next index with an unlocked
keep_simulating() + increment(), so near the end of a run two workers
could both pass the count < n check and then claim sim_idx == n; the
per-index child_seeds lookup turned that into an IndexError (before, it
only wrote one extra record). Move the claim into a _claim_next_index
helper that holds the shared mutex across the check and the increment,
so each index is handed out once and the counter never overshoots. A
deterministic unit test (a barrier plus a widened check-to-increment
window) over-claims and fails if the lock is dropped.

__root_seed_sequence returned the caller's SeedSequence, and spawn()
advances its child counter, so passing the same object to simulate()
twice produced different children. Copy it from its full state instead,
which leaves the caller untouched and keeps repeated calls reproducible.
Also drop Generator/BitGenerator from the accepted types: a stateful
generator is not a seed, and reducing it to its underlying SeedSequence
ignores how far it has been consumed. random_seed now takes an int, a
sequence of ints, or a SeedSequence.

Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com>

Author: thc1006

CHANGELOG.md

@@ -33,6 +33,8 @@ Attention: The newest changes should be on top -->
 ### Added
 
 - ENH: BUG/MNT: pre-release v1.13.0 review fixes [#1074](https://github.com/RocketPy-Team/RocketPy/pull/1074)
+- ENH: reproducible Monte Carlo runs via a random_seed argument [#1054](https://github.com/RocketPy-Team/RocketPy/pull/1054)
+
 ### Changed
 
 ### Fixed

rocketpy/simulation/monte_carlo.py

@@ -191,15 +191,15 @@ def simulate(
             number of workers will be equal to the number of CPUs available.
             A minimum of 2 workers is required for parallel mode.
             Default is None.
-        random_seed : int, numpy.random.SeedSequence, numpy.random.Generator, optional
+        random_seed : int or numpy.random.SeedSequence, optional
             Root seed for the run. When provided, the sampled inputs are
             reproducible and identical across serial and parallel execution
             and across any number of workers, because each simulation index
-            draws from its own child stream spawned from this root
-            (``SeedSequence(random_seed).spawn(number_of_simulations)``). It
-            accepts an int, a ``SeedSequence`` or a ``Generator``. Default is
-            None, which draws fresh entropy (not reproducible), preserving the
-            previous behavior.
+            draws from its own child stream spawned from this root. It accepts
+            an int or a ``SeedSequence``; a supplied ``SeedSequence`` is copied
+            rather than consumed, so repeated calls with the same seed produce
+            the same inputs. Default is None, which draws fresh entropy (not
+            reproducible), preserving the previous behavior.
         kwargs : dict
             Custom arguments for simulation export of the ``inputs`` file. Options
             are:
@@ -280,19 +280,26 @@ def __setup_files(self, append):
 
     @staticmethod
     def __root_seed_sequence(random_seed):
-        """Return a ``SeedSequence`` root from a flexible seed argument.
-
-        Accepts what the scientific-Python SPEC 7 seeding convention accepts
-        (an int, a ``SeedSequence``, a ``Generator`` or ``BitGenerator``, or
-        None for fresh entropy) and returns a ``SeedSequence`` so it can be
-        spawned into one independent child stream per simulation index.
+        """Build a fresh ``SeedSequence`` root from ``random_seed``.
+
+        ``random_seed`` may be an int (or any entropy ``numpy.random.SeedSequence``
+        accepts), an existing ``SeedSequence``, or ``None`` for fresh entropy. A
+        supplied ``SeedSequence`` is copied from its full ``state``, so the
+        spawning below neither mutates the caller's object nor advances a shared
+        child counter between calls; repeated ``simulate`` calls with the same
+        seed then stay reproducible. A stateful ``Generator``/``BitGenerator`` is
+        not accepted, since using it as an immutable seed would contradict its
+        consume-on-use semantics; pass ``rng.bit_generator.seed_seq`` to seed
+        from an existing generator's stream.
         """
         if isinstance(random_seed, np.random.SeedSequence):
-            return random_seed
-        if isinstance(random_seed, np.random.Generator):
-            return random_seed.bit_generator.seed_seq
-        if isinstance(random_seed, np.random.BitGenerator):
-            return random_seed.seed_seq
+            return np.random.SeedSequence(**random_seed.state)
+        if isinstance(random_seed, (np.random.Generator, np.random.BitGenerator)):
+            raise TypeError(
+                "random_seed must be an int or a numpy.random.SeedSequence, not "
+                f"a {type(random_seed).__name__}; to seed from an existing "
+                "generator pass rng.bit_generator.seed_seq."
+            )
         return np.random.SeedSequence(random_seed)
 
     def __seed_simulation(self, child_seed):
@@ -315,7 +322,7 @@ def __run_in_serial(self, random_seed=None):  # pylint: disable=too-many-stateme
 
         Parameters
         ----------
-        random_seed : int, SeedSequence, Generator, optional
+        random_seed : int or SeedSequence, optional
             Root seed for the run. See ``simulate``.
 
         Returns
@@ -369,7 +376,7 @@ def __run_in_parallel(self, n_workers=None, random_seed=None):
         n_workers: int, optional
             Number of workers to be used. If None, the number of workers
             will be equal to the number of CPUs available. Default is None.
-        random_seed : int, SeedSequence, Generator, optional
+        random_seed : int or SeedSequence, optional
             Root seed for the run. See ``simulate``.
 
         Returns
@@ -464,8 +471,11 @@ def __sim_producer(self, child_seeds, sim_monitor, mutex, error_event):  # pylin
             Event signaling an error occurred during the simulation.
         """
         try:
-            while sim_monitor.keep_simulating():
-                sim_idx = sim_monitor.increment() - 1
+            while True:
+                sim_idx = _claim_next_index(sim_monitor, mutex)
+                if sim_idx is None:
+                    break
+
                 inputs_json, outputs_json = "", ""
 
                 self.__seed_simulation(child_seeds[sim_idx])
@@ -1665,6 +1675,24 @@ def export_errors_to_json(self, filename):
         self._write_log_to_json(self.errors_log, filename)
 
 
+def _claim_next_index(sim_monitor, mutex):
+    """Atomically claim the next 0-based simulation index, or ``None`` if done.
+
+    ``keep_simulating()`` and ``increment()`` are two separate manager calls, so
+    the shared ``mutex`` has to be held across both. Without it, two workers can
+    each pass the ``count < number_of_simulations`` check at the tail before
+    either increments, and both then claim an index, overrunning the requested
+    number of simulations and indexing past the per-index seed list.
+    """
+    mutex.acquire()
+    try:
+        if not sim_monitor.keep_simulating():
+            return None
+        return sim_monitor.increment() - 1
+    finally:
+        mutex.release()
+
+
 def _import_multiprocess():
     """Import the necessary modules and submodules for the
     multiprocess library.

tests/unit/simulation/test_monte_carlo_determinism.py

@@ -4,9 +4,9 @@
 turning the run's root seed into one independent child stream per simulation
 index. Two private helpers do the work:
 
-* ``__root_seed_sequence`` normalizes the flexible ``random_seed`` argument (int,
-  ``SeedSequence``, ``Generator``, ``BitGenerator`` or None) into a
-  ``SeedSequence`` that can be spawned;
+* ``__root_seed_sequence`` normalizes the ``random_seed`` argument (an int, a
+  sequence of ints, a ``SeedSequence`` or None) into a fresh ``SeedSequence``
+  that can be spawned;
 * ``__seed_simulation`` splits one per-index child seed three ways so the
   environment, rocket and flight draw from independent streams.
 
@@ -19,12 +19,15 @@
 it lets the seeding invariants be asserted without running a Monte Carlo.
 """
 
+import threading
+import time
 from types import SimpleNamespace
 
 import numpy as np
 import pytest
 
 from rocketpy.simulation import MonteCarlo
+from rocketpy.simulation.monte_carlo import _SimMonitor, _claim_next_index
 
 _root_seed_sequence = MonteCarlo._MonteCarlo__root_seed_sequence
 _seed_simulation = MonteCarlo._MonteCarlo__seed_simulation
@@ -44,17 +47,17 @@ def _entropy(seed_sequence, n=4):
     "make_seed",
     [
         pytest.param(lambda: 12345, id="int"),
+        pytest.param(lambda: np.int64(12345), id="numpy-int"),
+        pytest.param(lambda: [1, 2, 3], id="sequence"),
         pytest.param(lambda: np.random.SeedSequence(12345), id="seedsequence"),
-        pytest.param(lambda: np.random.default_rng(12345), id="generator"),
-        pytest.param(lambda: np.random.PCG64(12345), id="bitgenerator"),
     ],
 )
-def test_root_seed_sequence_accepts_supported_types(make_seed):
-    """int, SeedSequence, Generator and BitGenerator all normalize to the same
-    root SeedSequence stream for an equivalent seed value."""
+def test_root_seed_sequence_accepts_seed_like_values(make_seed):
+    """An int, a numpy integer, a sequence of ints and a SeedSequence are all
+    accepted, normalize to a SeedSequence, and are reproducible."""
     root = _root_seed_sequence(make_seed())
     assert isinstance(root, np.random.SeedSequence)
-    assert _entropy(root) == _entropy(_root_seed_sequence(12345))
+    assert _entropy(root) == _entropy(_root_seed_sequence(make_seed()))
 
 
 def test_root_seed_sequence_none_draws_fresh_entropy():
@@ -65,30 +68,42 @@ def test_root_seed_sequence_none_draws_fresh_entropy():
 
 
 @pytest.mark.parametrize(
-    "make_seed, resolve",
+    "make_generator",
     [
-        pytest.param(
-            lambda: np.random.SeedSequence(999),
-            lambda seed: seed,
-            id="seedsequence",
-        ),
-        pytest.param(
-            lambda: np.random.default_rng(999),
-            lambda seed: seed.bit_generator.seed_seq,
-            id="generator",
-        ),
-        pytest.param(
-            lambda: np.random.PCG64(999),
-            lambda seed: seed.seed_seq,
-            id="bitgenerator",
-        ),
+        pytest.param(lambda: np.random.default_rng(999), id="generator"),
+        pytest.param(lambda: np.random.PCG64(999), id="bitgenerator"),
     ],
 )
-def test_root_seed_sequence_reuses_existing_seed_sequence(make_seed, resolve):
-    """When given something that already carries a SeedSequence, the helper
-    reuses that object rather than copying it."""
-    seed = make_seed()
-    assert _root_seed_sequence(seed) is resolve(seed)
+def test_root_seed_sequence_rejects_stateful_generators(make_generator):
+    """A Generator/BitGenerator is a stateful RNG, not a seed value, so it is
+    rejected instead of being reduced to its underlying SeedSequence."""
+    with pytest.raises(TypeError, match="SeedSequence"):
+        _root_seed_sequence(make_generator())
+
+
+def test_root_seed_sequence_copies_seedsequence_without_mutating_it():
+    """A supplied SeedSequence is copied from its full state: repeated calls with
+    the same object reproduce the same children, the caller's spawn counter is
+    left untouched, and a spawned child (non-empty spawn_key) round-trips too."""
+
+    def children(seed_sequence):
+        return [
+            _entropy(child) for child in _root_seed_sequence(seed_sequence).spawn(3)
+        ]
+
+    # A SeedSequence that has already spawned children, so its counter is not 0.
+    seed = np.random.SeedSequence(2024)
+    seed.spawn(5)
+    counter_before = seed.n_children_spawned
+
+    assert children(seed) == children(seed), "same object twice must reproduce"
+    assert seed.n_children_spawned == counter_before, "caller must not be mutated"
+    assert _root_seed_sequence(seed) is not seed, "must return a copy, not the caller"
+
+    # A spawned child carries a non-empty spawn_key that the copy must preserve.
+    child = np.random.SeedSequence(2024).spawn(1)[0]
+    assert child.spawn_key != ()
+    assert children(child) == children(child)
 
 
 # --------------------------------------------------------------------------- #
@@ -138,3 +153,57 @@ def split(child):
         return [_entropy(env[0]), _entropy(rocket[0]), _entropy(flight[0])]
 
     assert split(np.random.SeedSequence(2024)) == split(np.random.SeedSequence(2024))
+
+
+# --------------------------------------------------------------------------- #
+# _claim_next_index: atomic hand-out of the next simulation index             #
+# --------------------------------------------------------------------------- #
+
+
+def test_claim_next_index_hands_out_each_index_once_under_contention():
+    """Holding the mutex across keep_simulating() and increment() must hand out
+    each index exactly once, even when every worker reaches the claim together.
+
+    A barrier releases all workers at once and a widened check-to-increment
+    window would let an unlocked claim run several workers past the count < n
+    check before any increments; the lock is what keeps the result to exactly
+    n_simulations indices (0..n-1, none repeated) and the counter from
+    overshooting.
+    """
+    n_simulations = 5
+    n_workers = 8
+    monitor = _SimMonitor(initial_count=0, n_simulations=n_simulations, start_time=0.0)
+
+    # Widen the window between the check and the increment so that, without the
+    # lock, several workers could pass count < n before any of them increments.
+    real_keep_simulating = monitor.keep_simulating
+
+    def slow_keep_simulating():
+        result = real_keep_simulating()
+        time.sleep(0.02)
+        return result
+
+    monitor.keep_simulating = slow_keep_simulating
+
+    mutex = threading.Lock()
+    barrier = threading.Barrier(n_workers)
+    claimed = []
+    claimed_lock = threading.Lock()
+
+    def worker():
+        barrier.wait()
+        while True:
+            index = _claim_next_index(monitor, mutex)
+            if index is None:
+                break
+            with claimed_lock:
+                claimed.append(index)
+
+    workers = [threading.Thread(target=worker) for _ in range(n_workers)]
+    for thread in workers:
+        thread.start()
+    for thread in workers:
+        thread.join()
+
+    assert sorted(claimed) == list(range(n_simulations))
+    assert monitor.count == n_simulations