From 5a572d3ce39c91f677d0a391be371257ee78e9ae Mon Sep 17 00:00:00 2001
From: thc1006 <84045975+thc1006@users.noreply.github.com>
Date: Tue, 18 Aug 2026 02:07:55 +0800
Subject: [PATCH 01/10] BUG: accept the seed type a Monte Carlo worker is
 handed

A parallel run spawns a SeedSequence per worker and passes it to
environment, rocket and flight. _sampler_seed then fed it to
SeedSequence(entropy=...), which takes an int or a sequence of ints,
so the first worker raised TypeError before drawing anything.

The call was reached only from the custom sampler reset until #1117
added the list-choice generator, which every model goes through. A real
two-worker run passes at d21abde6^ in 2.32s and does not finish on
develop: the worker's own error path raises UnboundLocalError on
inputs_json, so the parent never learns it died and the run hangs.

The children of one root share their entropy and differ by spawn_key,
so the value is folded through generate_state rather than read off
entropy, which would put every worker on one sampler stream. Nothing
is consumed, and an int or None seed keeps the stream it had.

The fold lives in rocketpy.tools, since the component streams and the
per-index seeding both need the same one and three copies would drift
on width and word order. _sampler_seed does its own final fold through
it as well rather than repeating the four lines.

Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com>
---
 rocketpy/stochastic/stochastic_model.py       | 20 ++++-
 rocketpy/tools.py                             | 11 +++
 .../test_monte_carlo_parallel_runs.py         | 32 ++++++++
 tests/unit/stochastic/test_seed_types.py      | 81 +++++++++++++++++++
 4 files changed, 140 insertions(+), 4 deletions(-)
 create mode 100644 tests/unit/simulation/test_monte_carlo_parallel_runs.py
 create mode 100644 tests/unit/stochastic/test_seed_types.py

diff --git a/rocketpy/stochastic/stochastic_model.py b/rocketpy/stochastic/stochastic_model.py
index d42fb76c5..1dadb2f01 100644
--- a/rocketpy/stochastic/stochastic_model.py
+++ b/rocketpy/stochastic/stochastic_model.py
@@ -8,7 +8,7 @@
 from rocketpy.mathutils.function import Function
 from rocketpy.stochastic.custom_sampler import CustomSampler
 
-from ..tools import get_distribution
+from ..tools import _seed_sequence_to_int, get_distribution
 
 
 def _names_as_spawn_key(input_names):
@@ -41,6 +41,18 @@ def _format_number(value):
     return f"array of shape {np.shape(value)}"
 
 
+def _seed_as_entropy(seed):
+    """A seed as something ``SeedSequence`` will take as entropy.
+
+    A parallel run is handed a ``SeedSequence``, which it will not take. Any
+    other seed goes through untouched, so the stream an int reaches stays where
+    it was.
+    """
+    if not isinstance(seed, np.random.SeedSequence):
+        return seed
+    return _seed_sequence_to_int(seed)
+
+
 def _sampler_seed(seed, input_names):
     """Derive a seed for one sampler, or for one group that shares a generator.
 
@@ -54,10 +66,10 @@ def _sampler_seed(seed, input_names):
     # Sorted here rather than trusting the caller, so a future call site cannot
     # give one group two different seeds by listing its members another way.
     root = np.random.SeedSequence(
-        entropy=seed, spawn_key=_names_as_spawn_key(tuple(sorted(input_names)))
+        entropy=_seed_as_entropy(seed),
+        spawn_key=_names_as_spawn_key(tuple(sorted(input_names))),
     )
-    words = root.generate_state(4, dtype=np.uint32)
-    return sum(int(word) << (32 * position) for position, word in enumerate(words))
+    return _seed_sequence_to_int(root)
 
 
 # TODO: Stop using assert in production code. Use exceptions instead.
diff --git a/rocketpy/tools.py b/rocketpy/tools.py
index 0d7f1a74e..7f31f3e19 100644
--- a/rocketpy/tools.py
+++ b/rocketpy/tools.py
@@ -1377,6 +1377,17 @@ def euler313_to_quaternions(phi, theta, psi):
     return e0, e1, e2, e3
 
 
+def _seed_sequence_to_int(seed_sequence):
+    """Returns a ``SeedSequence`` as the 128-bit ``int`` it can be rebuilt from.
+
+    Folded through ``generate_state`` rather than read off ``entropy``, since
+    the children of one root differ only by ``spawn_key``, and combined by
+    value so it does not depend on byte order.
+    """
+    words = seed_sequence.generate_state(4, dtype=np.uint32)
+    return sum(int(word) << (32 * position) for position, word in enumerate(words))
+
+
 def get_matplotlib_supported_file_endings():
     """Gets the file endings supported by matplotlib.
 
diff --git a/tests/unit/simulation/test_monte_carlo_parallel_runs.py b/tests/unit/simulation/test_monte_carlo_parallel_runs.py
new file mode 100644
index 000000000..4ab0be440
--- /dev/null
+++ b/tests/unit/simulation/test_monte_carlo_parallel_runs.py
@@ -0,0 +1,32 @@
+import pytest
+
+from rocketpy.simulation.monte_carlo import MonteCarlo
+
+
+@pytest.mark.parametrize("parallel", [False, True])
+def test_a_monte_carlo_run_finishes(
+    stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path, parallel
+):
+    # The parallel path hands each worker a SeedSequence rather than an int, and
+    # nothing else in the suite exercises that. A worker that dies on it is not
+    # reported, so this reads as a hang rather than as a failure.
+    #
+    # Built here rather than taken from the monte_carlo_calisto fixture, whose
+    # own filename is fixed, since `filename` is a plain attribute and the three
+    # working paths are settled when the object is constructed.
+    analysis = MonteCarlo(
+        filename=str(tmp_path / "study"),
+        environment=stochastic_environment,
+        rocket=stochastic_calisto,
+        flight=stochastic_flight,
+    )
+
+    analysis.simulate(
+        number_of_simulations=2,
+        append=False,
+        parallel=parallel,
+        n_workers=2 if parallel else None,
+    )
+
+    assert analysis.num_of_loaded_sims == 2
+    assert str(tmp_path) in str(analysis.output_file)
diff --git a/tests/unit/stochastic/test_seed_types.py b/tests/unit/stochastic/test_seed_types.py
new file mode 100644
index 000000000..ba3d42583
--- /dev/null
+++ b/tests/unit/stochastic/test_seed_types.py
@@ -0,0 +1,81 @@
+import numpy as np
+import pytest
+
+from rocketpy.stochastic.stochastic_model import (
+    _names_as_spawn_key,
+    _sampler_seed,
+)
+from rocketpy.tools import _seed_sequence_to_int
+
+
+def _a_worker_seed(index=0, workers=2):
+    # What MonteCarlo.__run_in_parallel spawns and hands to each worker, which
+    # passes it straight to environment/rocket/flight._set_stochastic.
+    return np.random.SeedSequence().spawn(workers)[index]
+
+
+def test_the_seed_type_a_worker_is_handed_is_accepted(stochastic_calisto):
+    stochastic_calisto._set_stochastic(_a_worker_seed())
+
+    stochastic_calisto.create_object()
+
+
+def test_a_parachute_derives_its_noise_seed_from_a_worker_seed(
+    stochastic_main_parachute,
+):
+    stochastic_main_parachute._set_stochastic(_a_worker_seed())
+
+    assert stochastic_main_parachute.create_object().noise[2] is not None
+
+
+def test_two_workers_do_not_share_a_sampler_stream():
+    first, second = np.random.SeedSequence(7).spawn(2)
+    # They come off one root, so they carry the same entropy and differ only in
+    # spawn_key. Reading the entropy alone would put both on one stream.
+    assert first.entropy == second.entropy
+
+    assert _sampler_seed(first, ("__list_choice__",)) != _sampler_seed(
+        second, ("__list_choice__",)
+    )
+
+
+def test_a_caller_seed_sequence_is_not_consumed():
+    root = np.random.SeedSequence(42)
+
+    _sampler_seed(root, ("__list_choice__",))
+
+    assert root.n_children_spawned == 0
+    assert root.spawn(1)[0].spawn_key == (0,)
+
+
+def test_the_same_seed_sequence_twice_gives_the_same_sampler_seed():
+    root = np.random.SeedSequence(42)
+
+    first = _sampler_seed(root, ("pressure_noise", "main"))
+    second = _sampler_seed(root, ("pressure_noise", "main"))
+
+    assert first == second
+
+
+@pytest.mark.parametrize("seed", [42, 7, [1, 2, 3]])
+@pytest.mark.parametrize("names", [("__list_choice__",), ("pressure_noise", "main")])
+def test_a_seed_that_is_not_a_sequence_reaches_numpy_untouched(seed, names):
+    # The control. Every fixed-seed baseline in the suite was recorded through
+    # this path, so anything but a SeedSequence has to arrive as it always did.
+    # Compared with the expression rather than with a recorded number, which
+    # would go red on a NumPy release instead of on a change of ours.
+    unchanged = np.random.SeedSequence(
+        entropy=seed, spawn_key=_names_as_spawn_key(tuple(sorted(names)))
+    )
+
+    assert _sampler_seed(seed, names) == _seed_sequence_to_int(unchanged)
+
+
+def test_no_seed_still_means_no_seed():
+    # None is left out above on purpose: it asks NumPy for fresh entropy, so
+    # two calls must not agree, and comparing one against another would be
+    # asserting the opposite of what an unseeded run promises.
+    first = _sampler_seed(None, ("__list_choice__",))
+    second = _sampler_seed(None, ("__list_choice__",))
+
+    assert first != second

From 204c6d6ab09cf9f6296007118b8a24bc37d2e86e Mon Sep 17 00:00:00 2001
From: thc1006 <84045975+thc1006@users.noreply.github.com>
Date: Tue, 18 Aug 2026 02:14:45 +0800
Subject: [PATCH 02/10] BUG: report a worker that fails before its first
 simulation

__sim_producer binds sim_idx and inputs_json inside the loop, and its
handler reads both. A worker that fails in the seeding above the loop,
or in the claim that opens it, reached the handler with neither name
assigned and died with UnboundLocalError instead of recording anything.

The parent learns a worker failed from error_event, which the handler
sets on its last line, so it was never reached either and the run
waited rather than stopping.

Both names are bound before the try now, and the message says worker
startup when no index was claimed rather than naming one that does not
exist. Binding only the name in the traceback is not enough: the
message then raises on the other one, which the tests cover.

Scoped to the parallel producer. The serial handler has the same
unbound inputs_json and is #1177's.

Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com>
---
 rocketpy/simulation/monte_carlo.py            |   9 +-
 .../test_monte_carlo_worker_reporting.py      | 128 ++++++++++++++++++
 2 files changed, 134 insertions(+), 3 deletions(-)
 create mode 100644 tests/unit/simulation/test_monte_carlo_worker_reporting.py

diff --git a/rocketpy/simulation/monte_carlo.py b/rocketpy/simulation/monte_carlo.py
index c2dcd4030..41e40bdfc 100644
--- a/rocketpy/simulation/monte_carlo.py
+++ b/rocketpy/simulation/monte_carlo.py
@@ -531,6 +531,10 @@ def __sim_producer(self, seed, sim_monitor, mutex, error_event):  # pylint: disa
         error_event : multiprocess.Event
             Event signaling an error occurred during the simulation.
         """
+        # Bound before the try: the handler below reads both, and a failure in
+        # the seeding, or in the claim that opens the loop, reaches it with
+        # neither of them assigned.
+        sim_idx, inputs_json = None, ""
         try:
             # Ensure Processes generate different random numbers
             self.environment._set_stochastic(seed)
@@ -574,9 +578,8 @@ def __sim_producer(self, seed, sim_monitor, mutex, error_event):  # pylint: disa
 
             # See note above: must use print() to remain visible from a
             # multiprocessing worker process.
-            _SimMonitor.reprint(
-                f"Error on iteration {sim_idx}:\n{traceback.format_exc()}"
-            )
+            where = "worker startup" if sim_idx is None else f"iteration {sim_idx}"
+            _SimMonitor.reprint(f"Error on {where}:\n{traceback.format_exc()}")
             error_event.set()
             mutex.release()
 
diff --git a/tests/unit/simulation/test_monte_carlo_worker_reporting.py b/tests/unit/simulation/test_monte_carlo_worker_reporting.py
new file mode 100644
index 000000000..260ea48b8
--- /dev/null
+++ b/tests/unit/simulation/test_monte_carlo_worker_reporting.py
@@ -0,0 +1,128 @@
+import os
+from types import SimpleNamespace
+
+import pytest
+
+from rocketpy.simulation.monte_carlo import MonteCarlo
+
+
+class _Mutex:
+    def acquire(self):
+        pass
+
+    def release(self):
+        pass
+
+
+class _ErrorEvent:
+    def __init__(self):
+        self.was_set = False
+
+    def is_set(self):
+        return self.was_set
+
+    def set(self):
+        self.was_set = True
+
+
+def _a_worker(tmp_path, model):
+    study = MonteCarlo(
+        filename=os.path.join(str(tmp_path), "study"),
+        environment=model,
+        rocket=model,
+        flight=model,
+    )
+    return study, _ErrorEvent()
+
+
+def _run(study, monitor, error_event, seed=42):
+    # Name-mangled: the producer is what each worker process runs, and nothing
+    # else in the suite calls it.
+    study._MonteCarlo__sim_producer(seed, monitor, _Mutex(), error_event)
+
+
+def test_a_worker_that_fails_before_seeding_finishes_says_so(tmp_path, capsys):
+    def refuse(_seed):
+        raise RuntimeError("the models would not reseed")
+
+    model = SimpleNamespace(last_rnd_dict={}, _set_stochastic=refuse)
+    monitor = SimpleNamespace(keep_simulating=lambda: True)
+    study, error_event = _a_worker(tmp_path, model)
+
+    _run(study, monitor, error_event)
+
+    assert error_event.was_set
+    reported = capsys.readouterr().out
+    assert "worker startup" in reported
+    assert "the models would not reseed" in reported
+
+
+def test_a_worker_that_fails_before_claiming_an_index_says_so(tmp_path, capsys):
+    def refuse():
+        raise RuntimeError("the monitor would not hand out an index")
+
+    model = SimpleNamespace(last_rnd_dict={}, _set_stochastic=lambda _seed: None)
+    monitor = SimpleNamespace(keep_simulating=lambda: True, increment=refuse)
+    study, error_event = _a_worker(tmp_path, model)
+
+    _run(study, monitor, error_event)
+
+    assert error_event.was_set
+    assert "worker startup" in capsys.readouterr().out
+
+
+def test_a_worker_that_fails_inside_a_simulation_names_the_index(
+    tmp_path, capsys, monkeypatch
+):
+    # The control. An index is claimed and the simulation then fails, which is
+    # the path that already worked, so the report still has to name it.
+    def refuse(_self):
+        raise RuntimeError("the simulation would not run")
+
+    monkeypatch.setattr(
+        MonteCarlo, "_MonteCarlo__run_single_simulation", refuse, raising=True
+    )
+    model = SimpleNamespace(last_rnd_dict={}, _set_stochastic=lambda _seed: None)
+    monitor = SimpleNamespace(keep_simulating=lambda: True, increment=lambda: 8)
+    study, error_event = _a_worker(tmp_path, model)
+
+    _run(study, monitor, error_event)
+
+    assert error_event.was_set
+    assert "iteration 7" in capsys.readouterr().out
+
+
+def test_the_error_file_is_left_alone_when_nothing_was_drawn(tmp_path):
+    def refuse(_seed):
+        raise RuntimeError("the models would not reseed")
+
+    model = SimpleNamespace(last_rnd_dict={}, _set_stochastic=refuse)
+    monitor = SimpleNamespace(keep_simulating=lambda: True)
+    study, error_event = _a_worker(tmp_path, model)
+
+    _run(study, monitor, error_event)
+
+    with open(study.error_file, "r", encoding="utf-8") as recorded:
+        assert recorded.read() == ""
+
+
+@pytest.mark.parametrize("failing", ["_set_stochastic", "increment"])
+def test_a_worker_failure_never_raises_out_of_the_producer(tmp_path, failing):
+    # The handler used to reach for names the loop had not bound yet, so the
+    # process died with UnboundLocalError and the parent waited forever.
+    def refuse(*_args):
+        raise RuntimeError("boom")
+
+    model = SimpleNamespace(
+        last_rnd_dict={},
+        _set_stochastic=refuse if failing == "_set_stochastic" else lambda _s: None,
+    )
+    monitor = SimpleNamespace(
+        keep_simulating=lambda: True,
+        increment=refuse if failing == "increment" else (lambda: 1),
+    )
+    study, error_event = _a_worker(tmp_path, model)
+
+    _run(study, monitor, error_event)
+
+    assert error_event.was_set

From e32480f9d7532b4c5a5a025b95ec7d4d6a1c87f7 Mon Sep 17 00:00:00 2001
From: thc1006 <84045975+thc1006@users.noreply.github.com>
Date: Tue, 18 Aug 2026 02:47:20 +0800
Subject: [PATCH 03/10] BUG: notice a worker that never finished

The workers say they failed by setting an event, and the parent joins
them and reads it. A worker that is killed runs no handler, so the
event stays clear, the join returns because the process is gone, and
the run reports the simulations it never wrote as done.

Measured with a worker leaving in the second simulation of six, two
workers: simulate() returned normally with two rows on disk.

Its exit code is what is left of a worker that ends this way, so the
parent reads that too. Anything other than zero is refused, None
included, since that is a worker that has not finished at all.

The handler around it holds the manager mutex while it reports, so a
failure in the reporting left the lock held by a process that had
already gone and the next worker waited on it. The event is set first
and outside the lock, the lock is released from a finally, and each
reporting step is separate so an unwritable log cannot replace the
failure being reported. A startup failure writes a row of its own now
rather than nothing, since the caller is told to read that file.

The test leaves through the data collector rather than a patched
method, since a spawn platform re-imports the module in the child and
never sees the patch, and through os._exit rather than a signal, since
SIGKILL is POSIX-only. Checked on both start methods. The Monte Carlo
objects are built on tmp_path rather than retargeted, because filename
is a plain attribute and the three log paths are set in __init__.

Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com>
---
 rocketpy/simulation/monte_carlo.py            |  67 ++++++++++--
 .../test_monte_carlo_worker_exit.py           |  72 +++++++++++++
 .../test_monte_carlo_worker_reporting.py      | 101 ++++++++++++++----
 3 files changed, 211 insertions(+), 29 deletions(-)
 create mode 100644 tests/unit/simulation/test_monte_carlo_worker_exit.py

diff --git a/rocketpy/simulation/monte_carlo.py b/rocketpy/simulation/monte_carlo.py
index 41e40bdfc..e126dc0b6 100644
--- a/rocketpy/simulation/monte_carlo.py
+++ b/rocketpy/simulation/monte_carlo.py
@@ -18,6 +18,7 @@
 import os
 import traceback
 import warnings
+from contextlib import suppress
 from numbers import Real
 from pathlib import Path
 from time import time
@@ -488,6 +489,11 @@ def __run_in_parallel(self, n_workers=None):
                 for sim_producer in processes:
                     sim_producer.join()
 
+                # Before the event: a worker that was killed, or that died
+                # before its own handler could set it, leaves it clear, and the
+                # run would report the simulations it never wrote as done.
+                _refuse_a_worker_that_did_not_finish(processes)
+
                 # Handle error from the child processes
                 if simulation_error_event.is_set():
                     raise RuntimeError(
@@ -572,16 +578,29 @@ def __sim_producer(self, seed, sim_monitor, mutex, error_event):  # pylint: disa
                     mutex.release()
 
         except Exception:  # pylint: disable=broad-except
-            mutex.acquire()
-            with open(self.error_file, "a", encoding="utf-8") as f:
-                f.write(inputs_json)
-
-            # See note above: must use print() to remain visible from a
-            # multiprocessing worker process.
+            details = traceback.format_exc()
             where = "worker startup" if sim_idx is None else f"iteration {sim_idx}"
-            _SimMonitor.reprint(f"Error on {where}:\n{traceback.format_exc()}")
-            error_event.set()
-            mutex.release()
+            # Said first, and from outside the lock: a worker that cannot write
+            # its own diagnostics still has to be able to stop the others.
+            with suppress(Exception):
+                error_event.set()
+
+            mutex.acquire()
+            try:
+                # Suppressed, and every step separately: a full disk or an
+                # unwritable log would otherwise replace the failure being
+                # reported, and the lock is a manager's, so a worker that ends
+                # while holding it leaves the next one waiting on a process
+                # that no longer exists.
+                with suppress(Exception):
+                    with open(self.error_file, "a", encoding="utf-8") as f:
+                        f.write(inputs_json or _worker_failure_record(where, details))
+                with suppress(Exception):
+                    # See note above: must use print() to remain visible from a
+                    # multiprocessing worker process.
+                    _SimMonitor.reprint(f"Error on {where}:\n{details}")
+            finally:
+                mutex.release()
 
     def __run_single_simulation(self):
         """Runs a single simulation and returns the inputs and outputs.
@@ -1758,6 +1777,36 @@ def export_errors_to_json(self, filename):
         self._write_log_to_json(self.errors_log, filename)
 
 
+def _worker_failure_record(where, details):
+    """A row for a worker that failed before it drew anything.
+
+    Written because the caller is told to read the error file, and a traceback
+    a worker printed is not there to be read once its output is redirected.
+    """
+    return json.dumps({"index": None, "stage": where, "error": details}) + "\n"
+
+
+def _refuse_a_worker_that_did_not_finish(processes):
+    """Raise if any worker left without exiting cleanly.
+
+    The workers report their own failures through an event, which one that was
+    killed never reaches, so what is left of it is its exit code. A negative
+    one is the signal that ended it, and ``None`` is one still running.
+    """
+    unfinished = [
+        f"worker {position} with exit code {process.exitcode}"
+        for position, process in enumerate(processes)
+        if process.exitcode != 0
+    ]
+    if not unfinished:
+        return
+    raise RuntimeError(
+        f"The run is incomplete: {', '.join(unfinished)}. A worker that ends "
+        "this way records nothing and cannot say why, so the simulations it "
+        "held are missing from the results."
+    )
+
+
 def _import_multiprocess():
     """Import the necessary modules and submodules for the
     multiprocess library.
diff --git a/tests/unit/simulation/test_monte_carlo_worker_exit.py b/tests/unit/simulation/test_monte_carlo_worker_exit.py
new file mode 100644
index 000000000..d9ca0dc68
--- /dev/null
+++ b/tests/unit/simulation/test_monte_carlo_worker_exit.py
@@ -0,0 +1,72 @@
+import os
+from types import SimpleNamespace
+
+import pytest
+
+from rocketpy.simulation.monte_carlo import (
+    MonteCarlo,
+    _refuse_a_worker_that_did_not_finish,
+)
+
+
+def _worker(exitcode):
+    return SimpleNamespace(exitcode=exitcode)
+
+
+def test_workers_that_all_exited_cleanly_are_accepted():
+    _refuse_a_worker_that_did_not_finish([_worker(0), _worker(0)])
+
+
+def test_a_worker_killed_by_a_signal_is_refused():
+    with pytest.raises(RuntimeError, match=r"worker 1 with exit code -9"):
+        _refuse_a_worker_that_did_not_finish([_worker(0), _worker(-9)])
+
+
+def test_a_worker_that_exited_nonzero_is_refused():
+    with pytest.raises(RuntimeError, match=r"worker 0 with exit code 1"):
+        _refuse_a_worker_that_did_not_finish([_worker(1), _worker(0)])
+
+
+def test_every_unfinished_worker_is_named():
+    with pytest.raises(RuntimeError) as raised:
+        _refuse_a_worker_that_did_not_finish([_worker(-9), _worker(0), _worker(3)])
+
+    assert "worker 0" in str(raised.value)
+    assert "worker 2" in str(raised.value)
+    assert "worker 1" not in str(raised.value)
+
+
+@pytest.mark.parametrize("exitcode", [None, -15, 2])
+def test_anything_but_a_clean_exit_is_refused(exitcode):
+    with pytest.raises(RuntimeError):
+        _refuse_a_worker_that_did_not_finish([_worker(exitcode), _worker(0)])
+
+
+def _leave_without_recording(flight):  # pylint: disable=unused-argument
+    """Ends the worker the way a kill or an out-of-memory exit does.
+
+    ``os._exit`` rather than a signal, since ``SIGKILL`` is POSIX-only, and
+    reached through the data collector rather than a patched method, since a
+    ``spawn`` platform re-imports the module and would not see the patch.
+    """
+    os._exit(1)
+
+
+def test_a_worker_that_leaves_early_does_not_pass_as_a_finished_run(
+    stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path
+):
+    # The event the workers report through is set by their own handler, and
+    # this one leaves without running it, so the run used to return as though
+    # it had done every simulation it was asked for.
+    analysis = MonteCarlo(
+        filename=str(tmp_path / "study"),
+        environment=stochastic_environment,
+        rocket=stochastic_calisto,
+        flight=stochastic_flight,
+        data_collector={"leave": _leave_without_recording},
+    )
+
+    with pytest.raises(RuntimeError, match="incomplete"):
+        analysis.simulate(
+            number_of_simulations=6, append=False, parallel=True, n_workers=2
+        )
diff --git a/tests/unit/simulation/test_monte_carlo_worker_reporting.py b/tests/unit/simulation/test_monte_carlo_worker_reporting.py
index 260ea48b8..cf9d51ebc 100644
--- a/tests/unit/simulation/test_monte_carlo_worker_reporting.py
+++ b/tests/unit/simulation/test_monte_carlo_worker_reporting.py
@@ -1,53 +1,75 @@
+import json
 import os
 from types import SimpleNamespace
 
 import pytest
 
+from rocketpy.simulation import monte_carlo as mc_module
 from rocketpy.simulation.monte_carlo import MonteCarlo
 
 
 class _Mutex:
+    def __init__(self):
+        self.held = False
+        self.acquired = 0
+
     def acquire(self):
-        pass
+        self.acquired += 1
+        self.held = True
 
     def release(self):
-        pass
+        self.held = False
 
 
 class _ErrorEvent:
-    def __init__(self):
+    def __init__(self, refuse=False):
         self.was_set = False
+        self.refuse = refuse
 
     def is_set(self):
         return self.was_set
 
     def set(self):
+        if self.refuse:
+            raise OSError("the manager is gone")
         self.was_set = True
 
 
-def _a_worker(tmp_path, model):
+def _raise_instead(message):
+    def refuse(*_args, **_kwargs):
+        raise OSError(message)
+
+    return refuse
+
+
+def _refusing_model():
+    def refuse(_seed):
+        raise RuntimeError("the models would not reseed")
+
+    return SimpleNamespace(last_rnd_dict={}, _set_stochastic=refuse)
+
+
+def _a_worker(tmp_path, model, event=None):
     study = MonteCarlo(
-        filename=os.path.join(str(tmp_path), "study"),
+        filename=str(tmp_path / "study"),
         environment=model,
         rocket=model,
         flight=model,
     )
-    return study, _ErrorEvent()
+    return study, event or _ErrorEvent()
 
 
-def _run(study, monitor, error_event, seed=42):
+def _run(study, monitor, error_event, mutex=None):
     # Name-mangled: the producer is what each worker process runs, and nothing
     # else in the suite calls it.
-    study._MonteCarlo__sim_producer(seed, monitor, _Mutex(), error_event)
+    mutex = mutex or _Mutex()
+    study._MonteCarlo__sim_producer(42, monitor, mutex, error_event)
+    return mutex
 
 
 def test_a_worker_that_fails_before_seeding_finishes_says_so(tmp_path, capsys):
-    def refuse(_seed):
-        raise RuntimeError("the models would not reseed")
-
-    model = SimpleNamespace(last_rnd_dict={}, _set_stochastic=refuse)
     monitor = SimpleNamespace(keep_simulating=lambda: True)
-    study, error_event = _a_worker(tmp_path, model)
+    study, error_event = _a_worker(tmp_path, _refusing_model())
 
     _run(study, monitor, error_event)
 
@@ -92,18 +114,20 @@ def refuse(_self):
     assert "iteration 7" in capsys.readouterr().out
 
 
-def test_the_error_file_is_left_alone_when_nothing_was_drawn(tmp_path):
-    def refuse(_seed):
-        raise RuntimeError("the models would not reseed")
-
-    model = SimpleNamespace(last_rnd_dict={}, _set_stochastic=refuse)
+def test_a_startup_failure_is_written_down_and_not_only_printed(tmp_path):
+    # The caller is told to read the error file, and a traceback the worker
+    # printed is not there to be read once its output has been redirected.
     monitor = SimpleNamespace(keep_simulating=lambda: True)
-    study, error_event = _a_worker(tmp_path, model)
+    study, error_event = _a_worker(tmp_path, _refusing_model())
 
     _run(study, monitor, error_event)
 
     with open(study.error_file, "r", encoding="utf-8") as recorded:
-        assert recorded.read() == ""
+        rows = [json.loads(line) for line in recorded if line.strip()]
+    assert len(rows) == 1
+    assert rows[0]["index"] is None
+    assert rows[0]["stage"] == "worker startup"
+    assert "the models would not reseed" in rows[0]["error"]
 
 
 @pytest.mark.parametrize("failing", ["_set_stochastic", "increment"])
@@ -126,3 +150,40 @@ def refuse(*_args):
     _run(study, monitor, error_event)
 
     assert error_event.was_set
+
+
+@pytest.mark.parametrize("breaking", ["error_file", "reprint", "event"])
+def test_reporting_a_failure_never_keeps_the_mutex(tmp_path, monkeypatch, breaking):
+    # The mutex is the manager's, so a worker that ends while holding it leaves
+    # the next one waiting on a process that is gone, and the parent never
+    # reaches the join that would have noticed.
+    if breaking == "error_file":
+        monkeypatch.setattr(
+            mc_module, "_worker_failure_record", _raise_instead("no disk")
+        )
+    if breaking == "reprint":
+        monkeypatch.setattr(
+            mc_module._SimMonitor, "reprint", _raise_instead("no stdout")
+        )
+    event = _ErrorEvent(refuse=breaking == "event")
+    monitor = SimpleNamespace(keep_simulating=lambda: True)
+    study, error_event = _a_worker(tmp_path, _refusing_model(), event)
+
+    mutex = _run(study, monitor, error_event)
+
+    assert mutex.acquired == 1
+    assert not mutex.held
+
+
+def test_a_reporting_failure_does_not_replace_the_simulation_failure(
+    tmp_path, monkeypatch, capsys
+):
+    monkeypatch.setattr(mc_module, "_worker_failure_record", _raise_instead("no disk"))
+    monitor = SimpleNamespace(keep_simulating=lambda: True)
+    study, error_event = _a_worker(tmp_path, _refusing_model())
+
+    _run(study, monitor, error_event)
+
+    assert error_event.was_set
+    assert "the models would not reseed" in capsys.readouterr().out
+    assert not os.path.getsize(study.error_file)

From f25c61ee30931771301894924ccfee61e7672375 Mon Sep 17 00:00:00 2001
From: thc1006 <84045975+thc1006@users.noreply.github.com>
Date: Tue, 18 Aug 2026 04:17:46 +0800
Subject: [PATCH 04/10] BUG: stop waiting on workers once one of them has died

The lock the workers share belongs to the manager and is not released
when the process holding it is killed. A sibling then blocks on a lock
nobody owns, and the parent, joining without a timeout, waits with it.
The exit code check the previous commit added is never reached, so the
one case it exists for is the one it cannot see.

The join polls now, and acts only when a worker has actually ended
badly. A run that is merely slow is never bounded: an exit code, not a
duration, is what says a worker is gone. The survivors are asked
through the event first, since one between simulations leaves with its
logs intact, and only the ones still running after that are ended.

Undoing this leaves every test in the new file red. Deciding on how
long a worker has taken instead of on how it ended leaves exactly one
red, which is the test that says a slow run must be left alone.

Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com>
---
 rocketpy/simulation/monte_carlo.py            |  48 ++++++++-
 .../test_monte_carlo_worker_join.py           | 101 ++++++++++++++++++
 2 files changed, 147 insertions(+), 2 deletions(-)
 create mode 100644 tests/unit/simulation/test_monte_carlo_worker_join.py

diff --git a/rocketpy/simulation/monte_carlo.py b/rocketpy/simulation/monte_carlo.py
index e126dc0b6..7505542bb 100644
--- a/rocketpy/simulation/monte_carlo.py
+++ b/rocketpy/simulation/monte_carlo.py
@@ -486,8 +486,7 @@ def __run_in_parallel(self, n_workers=None):
                 sim_producer.start()
 
             try:
-                for sim_producer in processes:
-                    sim_producer.join()
+                _join_the_workers(processes, simulation_error_event)
 
                 # Before the event: a worker that was killed, or that died
                 # before its own handler could set it, leaves it clear, and the
@@ -1777,6 +1776,51 @@ def export_errors_to_json(self, filename):
         self._write_log_to_json(self.errors_log, filename)
 
 
+# Short enough that a dead worker is noticed promptly, long enough that the
+# polling costs nothing over a run that takes hours.
+_JOIN_POLL_SECONDS = 0.2
+_SHUTDOWN_GRACE_SECONDS = 5.0
+
+
+def _ended_badly(worker):
+    """Whether a worker has stopped, and stopped for the wrong reason."""
+    return worker.exitcode not in (None, 0)
+
+
+def _stop_the_workers_still_running(processes, error_event, grace_period):
+    """Ask the rest to stop, then end the ones that cannot.
+
+    Asked first because a worker between simulations reads the event and leaves
+    with its logs intact. One blocked on a lock its dead sibling was holding
+    never reaches that check, and only ending it frees the run.
+    """
+    with suppress(Exception):
+        error_event.set()
+    deadline = time() + grace_period
+    for worker in processes:
+        worker.join(timeout=max(0.0, deadline - time()))
+    for worker in processes:
+        if worker.is_alive():
+            worker.terminate()
+            worker.join(timeout=grace_period)
+
+
+def _join_the_workers(processes, error_event, grace_period=_SHUTDOWN_GRACE_SECONDS):
+    """Wait for the workers, and stop waiting once one of them has died badly.
+
+    The lock the workers share belongs to the manager and is not released when
+    its holder is killed, so a sibling can block on a lock nobody owns while an
+    unbounded join waits with it. Nothing here bounds a run that is merely
+    slow: only an exit code says a worker has died.
+    """
+    while any(worker.is_alive() for worker in processes):
+        for worker in processes:
+            worker.join(timeout=_JOIN_POLL_SECONDS)
+        if any(_ended_badly(worker) for worker in processes):
+            _stop_the_workers_still_running(processes, error_event, grace_period)
+            return
+
+
 def _worker_failure_record(where, details):
     """A row for a worker that failed before it drew anything.
 
diff --git a/tests/unit/simulation/test_monte_carlo_worker_join.py b/tests/unit/simulation/test_monte_carlo_worker_join.py
new file mode 100644
index 000000000..7ed1253ad
--- /dev/null
+++ b/tests/unit/simulation/test_monte_carlo_worker_join.py
@@ -0,0 +1,101 @@
+import pytest
+
+from rocketpy.simulation.monte_carlo import _join_the_workers
+
+
+class _Worker:
+    """A process that stops after a set number of polls, or never.
+
+    ``never`` stands in for one blocked on a lock its dead sibling was holding,
+    which is the case an unbounded join waits out forever.
+    """
+
+    def __init__(self, exitcode=0, alive_for=0, never=False):
+        self.exitcode = None
+        self._final_exitcode = exitcode
+        self._alive_for = alive_for
+        self._never = never
+        self.joins = 0
+        self.terminated = False
+
+    def is_alive(self):
+        return self.exitcode is None
+
+    def join(self, timeout=None):  # pylint: disable=unused-argument
+        self.joins += 1
+        if self._never or self.joins <= self._alive_for:
+            return
+        self.exitcode = self._final_exitcode
+
+    def terminate(self):
+        self.terminated = True
+        self.exitcode = -15
+
+
+class _Event:
+    def __init__(self):
+        self.was_set = False
+
+    def set(self):
+        self.was_set = True
+
+
+def test_a_run_where_every_worker_finishes_is_left_alone():
+    workers = [_Worker(alive_for=3), _Worker(alive_for=5)]
+
+    _join_the_workers(workers, _Event(), grace_period=0)
+
+    assert [worker.exitcode for worker in workers] == [0, 0]
+    assert not any(worker.terminated for worker in workers)
+
+
+def test_a_worker_blocked_behind_a_dead_one_does_not_wait_forever():
+    # The one that mattered. Without a bound this call never returns, so the
+    # parent never reaches the check that would have reported the failure.
+    died = _Worker(exitcode=-9, alive_for=1)
+    blocked = _Worker(never=True)
+
+    _join_the_workers([died, blocked], _Event(), grace_period=0)
+
+    assert blocked.terminated
+
+
+def test_the_survivors_are_asked_before_they_are_ended():
+    died = _Worker(exitcode=1, alive_for=1)
+    blocked = _Worker(never=True)
+    event = _Event()
+
+    _join_the_workers([died, blocked], event, grace_period=0)
+
+    assert event.was_set
+
+
+def test_a_survivor_that_stops_on_its_own_is_not_terminated():
+    died = _Worker(exitcode=1, alive_for=1)
+    cooperative = _Worker(alive_for=2)
+
+    _join_the_workers([died, cooperative], _Event(), grace_period=0)
+
+    assert not cooperative.terminated
+    assert cooperative.exitcode == 0
+
+
+@pytest.mark.parametrize("exitcode", [-9, 1, 2])
+def test_any_bad_exit_starts_the_shutdown(exitcode):
+    died = _Worker(exitcode=exitcode, alive_for=1)
+    blocked = _Worker(never=True)
+
+    _join_the_workers([died, blocked], _Event(), grace_period=0)
+
+    assert blocked.terminated
+
+
+def test_a_slow_run_is_never_bounded():
+    # Nothing here may act on how long a worker takes, only on it having died.
+    slow = _Worker(alive_for=50)
+    slower = _Worker(alive_for=80)
+
+    _join_the_workers([slow, slower], _Event(), grace_period=0)
+
+    assert not any(worker.terminated for worker in (slow, slower))
+    assert slower.joins > 50

From 3a73caae80dfb55ee4b2ed9c4429db790b830ebe Mon Sep 17 00:00:00 2001
From: thc1006 <84045975+thc1006@users.noreply.github.com>
Date: Tue, 18 Aug 2026 06:10:26 +0800
Subject: [PATCH 05/10] MNT: give the worker failure report a name of its own

Lifted out of __sim_producer unchanged. The producer was over pylint's
statement limit once the per-index seeding shortens it elsewhere, and
the handler is one thing rather than part of the loop around it.

Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com>
---
 rocketpy/simulation/monte_carlo.py | 48 ++++++++++++++++--------------
 1 file changed, 26 insertions(+), 22 deletions(-)

diff --git a/rocketpy/simulation/monte_carlo.py b/rocketpy/simulation/monte_carlo.py
index 7505542bb..af28b4a13 100644
--- a/rocketpy/simulation/monte_carlo.py
+++ b/rocketpy/simulation/monte_carlo.py
@@ -577,29 +577,33 @@ def __sim_producer(self, seed, sim_monitor, mutex, error_event):  # pylint: disa
                     mutex.release()
 
         except Exception:  # pylint: disable=broad-except
-            details = traceback.format_exc()
-            where = "worker startup" if sim_idx is None else f"iteration {sim_idx}"
-            # Said first, and from outside the lock: a worker that cannot write
-            # its own diagnostics still has to be able to stop the others.
-            with suppress(Exception):
-                error_event.set()
+            self.__report_a_failed_simulation(sim_idx, inputs_json, mutex, error_event)
 
-            mutex.acquire()
-            try:
-                # Suppressed, and every step separately: a full disk or an
-                # unwritable log would otherwise replace the failure being
-                # reported, and the lock is a manager's, so a worker that ends
-                # while holding it leaves the next one waiting on a process
-                # that no longer exists.
-                with suppress(Exception):
-                    with open(self.error_file, "a", encoding="utf-8") as f:
-                        f.write(inputs_json or _worker_failure_record(where, details))
-                with suppress(Exception):
-                    # See note above: must use print() to remain visible from a
-                    # multiprocessing worker process.
-                    _SimMonitor.reprint(f"Error on {where}:\n{details}")
-            finally:
-                mutex.release()
+    def __report_a_failed_simulation(self, sim_idx, inputs_json, mutex, error_event):
+        """Write down and announce a simulation this worker could not finish.
+
+        The event goes first and from outside the lock, since a worker that
+        cannot write its diagnostics still has to be able to stop the others.
+        Each step under the lock is suppressed on its own: a full disk would
+        otherwise replace the failure being reported, and the lock is a
+        manager's, so ending while holding it leaves the next worker waiting
+        on a process that no longer exists.
+        """
+        details = traceback.format_exc()
+        where = "worker startup" if sim_idx is None else f"iteration {sim_idx}"
+        with suppress(Exception):
+            error_event.set()
+
+        mutex.acquire()
+        try:
+            with suppress(Exception):
+                with open(self.error_file, "a", encoding="utf-8") as f:
+                    f.write(inputs_json or _worker_failure_record(where, details))
+            with suppress(Exception):
+                # Must use print() to remain visible from a worker process.
+                _SimMonitor.reprint(f"Error on {where}:\n{details}")
+        finally:
+            mutex.release()
 
     def __run_single_simulation(self):
         """Runs a single simulation and returns the inputs and outputs.

From 43428b4a6938b32c79e0e8c03cff77227793799a Mon Sep 17 00:00:00 2001
From: thc1006 <84045975+thc1006@users.noreply.github.com>
Date: Thu, 20 Aug 2026 16:40:18 +0800
Subject: [PATCH 06/10] BUG: judge a parallel run on its logs, not on how its
 workers ended

Three ways a failed run still passed for a finished one, all of them
found by review of the previous commits here.

The join waited only on exit status. A worker that fails the ordinary
way is caught by the producer, reports through the event and returns,
so it exits cleanly. With a sibling stuck, the parent saw one clean
exit and one live process and never stopped. A reported failure ends
the wait now as well, and a manager that cannot be asked is not taken
as evidence either way.

The bounded shutdown was undone one level up: after the exit-code
check raised, the outer handler joined every process again with no
timeout, so the stubborn worker it exists for was waited on anyway.
That handler uses the same bounded teardown now, and a test walks the
parallel path's syntax to keep an unbounded join out of it.

An exit code says how a process ended, never whether the index it had
claimed reached the logs, and the monitor counts claims rather than
rows. Measured, six simulations across two workers leaving through
os._exit(0): simulate() returned normally with nothing written. The
run is checked against the logs themselves at the end now. Both must
hold exactly the simulations asked for, none twice, none unreadable,
none numbered past the run.

The stand-in worker in the join tests gives up after two hundred
polls. Reproducing a real hang there would take a CI job down with it
rather than report, and requirements-tests.txt has no pytest-timeout.

Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com>
---
 rocketpy/simulation/monte_carlo.py            |  89 ++++++++++++-
 .../test_monte_carlo_run_completeness.py      | 123 ++++++++++++++++++
 .../test_monte_carlo_worker_join.py           |  48 ++++++-
 3 files changed, 252 insertions(+), 8 deletions(-)
 create mode 100644 tests/unit/simulation/test_monte_carlo_run_completeness.py

diff --git a/rocketpy/simulation/monte_carlo.py b/rocketpy/simulation/monte_carlo.py
index af28b4a13..9d982df10 100644
--- a/rocketpy/simulation/monte_carlo.py
+++ b/rocketpy/simulation/monte_carlo.py
@@ -501,15 +501,25 @@ def __run_in_parallel(self, n_workers=None):
                         "for more information."
                     )
 
+                # Last, and from the logs rather than from the workers: every
+                # check above reads how a process ended, and none of them can
+                # see a worker that left cleanly between claiming an index and
+                # recording it.
+                _refuse_a_run_that_lost_a_simulation(
+                    self.input_file, self.output_file, self.number_of_simulations
+                )
+
                 sim_monitor.print_final_status()
 
             # Handle error from the main process
             # pylint: disable=broad-except
             except (Exception, KeyboardInterrupt) as error:
-                simulation_error_event.set()
-
-                for sim_producer in processes:
-                    sim_producer.join()
+                # The same bounded teardown, which sets the event itself. An
+                # unbounded join here used to undo the bound above on exactly
+                # the stubborn worker it exists for.
+                _stop_the_workers_still_running(
+                    processes, simulation_error_event, _SHUTDOWN_GRACE_SECONDS
+                )
 
                 if not isinstance(error, KeyboardInterrupt):
                     raise error
@@ -1791,6 +1801,20 @@ def _ended_badly(worker):
     return worker.exitcode not in (None, 0)
 
 
+def _the_run_is_already_lost(processes, error_event):
+    """Whether anything says the run cannot finish.
+
+    A worker that fails the ordinary way reports through the event and returns,
+    so it exits cleanly and its exit code says nothing. Waiting only on exit
+    codes leaves the parent sitting behind a sibling that is stuck.
+    """
+    if any(_ended_badly(worker) for worker in processes):
+        return True
+    with suppress(Exception):
+        return bool(error_event.is_set())
+    return False
+
+
 def _stop_the_workers_still_running(processes, error_event, grace_period):
     """Ask the rest to stop, then end the ones that cannot.
 
@@ -1815,12 +1839,12 @@ def _join_the_workers(processes, error_event, grace_period=_SHUTDOWN_GRACE_SECON
     The lock the workers share belongs to the manager and is not released when
     its holder is killed, so a sibling can block on a lock nobody owns while an
     unbounded join waits with it. Nothing here bounds a run that is merely
-    slow: only an exit code says a worker has died.
+    slow: only an exit code or a reported failure says a worker has given up.
     """
     while any(worker.is_alive() for worker in processes):
         for worker in processes:
             worker.join(timeout=_JOIN_POLL_SECONDS)
-        if any(_ended_badly(worker) for worker in processes):
+        if _the_run_is_already_lost(processes, error_event):
             _stop_the_workers_still_running(processes, error_event, grace_period)
             return
 
@@ -1834,6 +1858,59 @@ def _worker_failure_record(where, details):
     return json.dumps({"index": None, "stage": where, "error": details}) + "\n"
 
 
+def _indices_a_log_holds(path):
+    """Every index a log records, in order, and ``None`` for a row it cannot."""
+    found = []
+    with open(path, "r", encoding="utf-8") as recorded:
+        for line in recorded:
+            if not line.strip():
+                continue
+            try:
+                found.append(json.loads(line)["index"])
+            except (ValueError, KeyError, TypeError):
+                found.append(None)
+    return found
+
+
+def _refuse_a_run_that_lost_a_simulation(input_file, output_file, target):
+    """Raise unless both logs hold every simulation the run was asked for.
+
+    An exit code says how a worker ended, never whether the index it had
+    already claimed reached the logs, and the monitor counts claims rather than
+    rows. A worker that leaves between the two is invisible to everything else
+    here, so the logs themselves are what the run is judged on.
+    """
+    wanted = set(range(target))
+    for label, path in (("input", input_file), ("output", output_file)):
+        found = _indices_a_log_holds(path)
+        held = set(found)
+        if None in held:
+            raise RuntimeError(
+                f"The run is incomplete: the {label} log has rows that cannot "
+                f"be read, so what it holds cannot be established."
+            )
+        if len(found) != len(held):
+            raise RuntimeError(
+                f"The run is incomplete: the {label} log records "
+                f"{len(found) - len(held)} simulation(s) more than once."
+            )
+        if held != wanted:
+            missing = sorted(wanted - held)
+            extra = sorted(held - wanted)
+            trouble = []
+            if missing:
+                trouble.append(
+                    f"{len(missing)} of {target} are missing, the first "
+                    f"being {missing[0]}"
+                )
+            if extra:
+                trouble.append(f"{len(extra)} are numbered past the run")
+            raise RuntimeError(
+                f"The run is incomplete: the {label} log does not hold every "
+                f"simulation that was asked for, {' and '.join(trouble)}."
+            )
+
+
 def _refuse_a_worker_that_did_not_finish(processes):
     """Raise if any worker left without exiting cleanly.
 
diff --git a/tests/unit/simulation/test_monte_carlo_run_completeness.py b/tests/unit/simulation/test_monte_carlo_run_completeness.py
new file mode 100644
index 000000000..005eb31d6
--- /dev/null
+++ b/tests/unit/simulation/test_monte_carlo_run_completeness.py
@@ -0,0 +1,123 @@
+import ast
+import inspect
+import json
+import os
+
+import pytest
+
+from rocketpy.simulation import monte_carlo as mc_module
+from rocketpy.simulation.monte_carlo import (
+    MonteCarlo,
+    _refuse_a_run_that_lost_a_simulation,
+)
+
+
+def _a_log(tmp_path, name, rows):
+    path = tmp_path / name
+    path.write_text("".join(rows), encoding="utf-8")
+    return str(path)
+
+
+def _row(index):
+    return json.dumps({"index": index, "mass": 1.0}) + "\n"
+
+
+def _complete(tmp_path, count=3, name="ok"):
+    rows = [_row(index) for index in range(count)]
+    return (
+        _a_log(tmp_path, f"{name}.inputs.txt", rows),
+        _a_log(tmp_path, f"{name}.outputs.txt", rows),
+    )
+
+
+def test_a_run_that_recorded_everything_is_accepted(tmp_path):
+    inputs, outputs = _complete(tmp_path)
+
+    _refuse_a_run_that_lost_a_simulation(inputs, outputs, 3)
+
+
+def test_a_missing_simulation_is_refused(tmp_path):
+    inputs, outputs = _complete(tmp_path)
+    _a_log(tmp_path, "ok.outputs.txt", [_row(0), _row(2)])
+
+    with pytest.raises(RuntimeError, match=r"output log.*missing.*being 1"):
+        _refuse_a_run_that_lost_a_simulation(inputs, outputs, 3)
+
+
+def test_a_simulation_recorded_twice_is_refused(tmp_path):
+    inputs, outputs = _complete(tmp_path)
+    _a_log(tmp_path, "ok.inputs.txt", [_row(0), _row(1), _row(1), _row(2)])
+
+    with pytest.raises(RuntimeError, match="more than once"):
+        _refuse_a_run_that_lost_a_simulation(inputs, outputs, 3)
+
+
+def test_a_row_that_cannot_be_read_is_refused(tmp_path):
+    inputs, outputs = _complete(tmp_path)
+    _a_log(tmp_path, "ok.outputs.txt", [_row(0), "{half a row\n", _row(2)])
+
+    with pytest.raises(RuntimeError, match="cannot be read"):
+        _refuse_a_run_that_lost_a_simulation(inputs, outputs, 3)
+
+
+def test_a_row_numbered_past_the_run_is_refused(tmp_path):
+    inputs, outputs = _complete(tmp_path)
+    _a_log(tmp_path, "ok.inputs.txt", [_row(0), _row(1), _row(2), _row(9)])
+
+    with pytest.raises(RuntimeError, match="past the run"):
+        _refuse_a_run_that_lost_a_simulation(inputs, outputs, 3)
+
+
+def test_logs_that_hold_different_simulations_are_refused(tmp_path):
+    inputs = _a_log(tmp_path, "a.inputs.txt", [_row(0), _row(1)])
+    outputs = _a_log(tmp_path, "a.outputs.txt", [_row(0), _row(2)])
+
+    with pytest.raises(RuntimeError):
+        _refuse_a_run_that_lost_a_simulation(inputs, outputs, 2)
+
+
+def _leave_cleanly_without_recording(_flight):
+    # A worker that ends the way an out-of-memory kill ends it, but with the
+    # status of one that finished. Nothing about the process says otherwise.
+    os._exit(0)
+
+
+def test_a_worker_that_leaves_cleanly_without_recording_is_not_a_success(
+    stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path
+):
+    analysis = MonteCarlo(
+        filename=str(tmp_path / "study"),
+        environment=stochastic_environment,
+        rocket=stochastic_calisto,
+        flight=stochastic_flight,
+        data_collector={"leave": _leave_cleanly_without_recording},
+    )
+
+    with pytest.raises(RuntimeError, match="incomplete"):
+        analysis.simulate(
+            number_of_simulations=6, append=False, parallel=True, n_workers=2
+        )
+
+
+def test_no_failure_path_waits_on_a_worker_without_a_bound():
+    # An unbounded join anywhere in the parallel path puts back the hang that
+    # the bounded teardown exists to end, and it does so where it is hardest
+    # to notice: only when a worker is already stuck.
+    tree = ast.parse(inspect.getsource(mc_module))
+    run_in_parallel = next(
+        node
+        for node in ast.walk(tree)
+        if isinstance(node, ast.FunctionDef) and node.name == "__run_in_parallel"
+    )
+
+    unbounded = [
+        node.lineno
+        for node in ast.walk(run_in_parallel)
+        if isinstance(node, ast.Call)
+        and isinstance(node.func, ast.Attribute)
+        and node.func.attr == "join"
+        and not node.args
+        and not node.keywords
+    ]
+
+    assert not unbounded, f"join() with no timeout at lines {unbounded}"
diff --git a/tests/unit/simulation/test_monte_carlo_worker_join.py b/tests/unit/simulation/test_monte_carlo_worker_join.py
index 7ed1253ad..20ef8df22 100644
--- a/tests/unit/simulation/test_monte_carlo_worker_join.py
+++ b/tests/unit/simulation/test_monte_carlo_worker_join.py
@@ -23,6 +23,10 @@ def is_alive(self):
 
     def join(self, timeout=None):  # pylint: disable=unused-argument
         self.joins += 1
+        # A real worker that never returns makes the caller hang, which is the
+        # bug. Reproducing that here would hang CI instead of reporting, so the
+        # stand-in gives up and says so.
+        assert self.joins < 200, "the join loop never stopped waiting"
         if self._never or self.joins <= self._alive_for:
             return
         self.exitcode = self._final_exitcode
@@ -33,13 +37,23 @@ def terminate(self):
 
 
 class _Event:
-    def __init__(self):
-        self.was_set = False
+    def __init__(self, already_set=False):
+        self.was_set = already_set
+
+    def is_set(self):
+        return self.was_set
 
     def set(self):
         self.was_set = True
 
 
+class _BrokenEvent(_Event):
+    """A manager proxy that has gone away."""
+
+    def is_set(self):
+        raise OSError("the manager is gone")
+
+
 def test_a_run_where_every_worker_finishes_is_left_alone():
     workers = [_Worker(alive_for=3), _Worker(alive_for=5)]
 
@@ -99,3 +113,33 @@ def test_a_slow_run_is_never_bounded():
 
     assert not any(worker.terminated for worker in (slow, slower))
     assert slower.joins > 50
+
+
+def test_a_reported_failure_also_stops_a_sibling_that_never_returns():
+    # A worker that fails the ordinary way is caught by the producer, reports
+    # through the event and returns, so it exits cleanly. Waiting only on exit
+    # codes leaves the parent sitting behind whichever sibling is stuck.
+    reported = _Worker(exitcode=0, alive_for=1)
+    stuck = _Worker(never=True)
+
+    _join_the_workers([reported, stuck], _Event(already_set=True), grace_period=0)
+
+    assert stuck.terminated
+
+
+def test_a_clean_run_is_not_stopped_by_an_event_nobody_set():
+    first, second = _Worker(alive_for=2), _Worker(alive_for=3)
+
+    _join_the_workers([first, second], _Event(), grace_period=0)
+
+    assert not any(worker.terminated for worker in (first, second))
+
+
+def test_an_event_that_cannot_be_read_does_not_stop_the_run():
+    # The control on the control. If asking the manager raises, that is not
+    # evidence of failure and must not end a healthy run.
+    first, second = _Worker(alive_for=2), _Worker(alive_for=3)
+
+    _join_the_workers([first, second], _BrokenEvent(), grace_period=0)
+
+    assert not any(worker.terminated for worker in (first, second))

From 7c3b03f35043af6d6909d2dc6d07315674b14b6f Mon Sep 17 00:00:00 2001
From: thc1006 <84045975+thc1006@users.noreply.github.com>
Date: Tue, 25 Aug 2026 22:53:38 +0800
Subject: [PATCH 07/10] BUG: report the simulation that failed, and bring the
 fleet down whole

Review of the previous commits found three things, and reviewing my own
answer to the first of them found a fourth.

A worker kept sim_idx and inputs_json after committing a row, so a
failure in the claim that opened the next round was reported against
the simulation that had just succeeded, and that simulation's inputs
were written to the error log as well. Measured: output held index 0
and the error log held index 0 with the same inputs. They are cleared
once a row is committed, since nothing is in flight between two
simulations.

A worker whose event could not be set returned anyway, so it exited
cleanly and the parent saw neither an event nor a bad exit. It re-
raises when it could not announce, leaving the exit code to say so.

The bounded shutdown deserved more than one stage. It shares one
deadline across the fleet rather than giving each worker its own, uses
a monotonic clock so a correction cannot move a shutdown already under
way, and kills what outlives terminate.

The fourth is mine. I had made a reported failure end the wait, which
sounded right and was not: a worker that merely reports leaves nothing
behind, and its siblings stop of their own accord once they finish the
simulation in hand. Measured, a sibling that needed forty polls was
terminated after three. Only a worker that died can hold the shared
lock for good, so only that ends the wait now.

The completeness check no longer refuses rows numbered past the target.
An append given a smaller target than the checkpoint holds is an append
question, and refusing it here both changed what append means and said
rows were missing when they were extra.

Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com>
---
 rocketpy/simulation/monte_carlo.py            | 120 ++++++++++--------
 .../test_monte_carlo_parallel_runs.py         |   1 +
 .../test_monte_carlo_run_completeness.py      |  32 +++--
 .../test_monte_carlo_worker_exit.py           |   8 +-
 .../test_monte_carlo_worker_join.py           |  74 +++++++----
 .../test_monte_carlo_worker_reporting.py      | 105 ++++++++++++++-
 6 files changed, 249 insertions(+), 91 deletions(-)

diff --git a/rocketpy/simulation/monte_carlo.py b/rocketpy/simulation/monte_carlo.py
index 9d982df10..50f95b23b 100644
--- a/rocketpy/simulation/monte_carlo.py
+++ b/rocketpy/simulation/monte_carlo.py
@@ -21,7 +21,7 @@
 from contextlib import suppress
 from numbers import Real
 from pathlib import Path
-from time import time
+from time import monotonic, time
 
 import numpy as np
 import simplekml
@@ -301,6 +301,14 @@ def simulate(
         -------
         None
 
+        Raises
+        ------
+        RuntimeError
+            If a parallel run does not finish. A worker that ends badly, one
+            that reports a failure, and logs that do not hold every simulation
+            asked for are each refused, since a run that lost work must not be
+            reported as one that completed.
+
         Notes
         -----
         If you need to stop the simulations after starting them, you can
@@ -488,9 +496,7 @@ def __run_in_parallel(self, n_workers=None):
             try:
                 _join_the_workers(processes, simulation_error_event)
 
-                # Before the event: a worker that was killed, or that died
-                # before its own handler could set it, leaves it clear, and the
-                # run would report the simulations it never wrote as done.
+                # Before the event: a killed worker never sets it.
                 _refuse_a_worker_that_did_not_finish(processes)
 
                 # Handle error from the child processes
@@ -501,11 +507,9 @@ def __run_in_parallel(self, n_workers=None):
                         "for more information."
                     )
 
-                # Last, and from the logs rather than from the workers: every
-                # check above reads how a process ended, and none of them can
-                # see a worker that left cleanly between claiming an index and
-                # recording it.
-                _refuse_a_run_that_lost_a_simulation(
+                # An exit code cannot show a worker that left between
+                # claiming an index and recording it.
+                _refuse_logs_missing_a_simulation(
                     self.input_file, self.output_file, self.number_of_simulations
                 )
 
@@ -514,9 +518,7 @@ def __run_in_parallel(self, n_workers=None):
             # Handle error from the main process
             # pylint: disable=broad-except
             except (Exception, KeyboardInterrupt) as error:
-                # The same bounded teardown, which sets the event itself. An
-                # unbounded join here used to undo the bound above on exactly
-                # the stubborn worker it exists for.
+                # Bounded here too. An unbounded join undid the bound above.
                 _stop_the_workers_still_running(
                     processes, simulation_error_event, _SHUTDOWN_GRACE_SECONDS
                 )
@@ -546,9 +548,7 @@ def __sim_producer(self, seed, sim_monitor, mutex, error_event):  # pylint: disa
         error_event : multiprocess.Event
             Event signaling an error occurred during the simulation.
         """
-        # Bound before the try: the handler below reads both, and a failure in
-        # the seeding, or in the claim that opens the loop, reaches it with
-        # neither of them assigned.
+        # The handler reads both, and a failure above the loop precedes them.
         sim_idx, inputs_json = None, ""
         try:
             # Ensure Processes generate different random numbers
@@ -586,8 +586,15 @@ def __sim_producer(self, seed, sim_monitor, mutex, error_event):  # pylint: disa
                 finally:
                     mutex.release()
 
+                # Nothing is in flight between two simulations, nor are these.
+                sim_idx, inputs_json = None, ""
+
         except Exception:  # pylint: disable=broad-except
-            self.__report_a_failed_simulation(sim_idx, inputs_json, mutex, error_event)
+            if not self.__report_a_failed_simulation(
+                sim_idx, inputs_json, mutex, error_event
+            ):
+                # The event could not be set; the exit code is what is left.
+                raise
 
     def __report_a_failed_simulation(self, sim_idx, inputs_json, mutex, error_event):
         """Write down and announce a simulation this worker could not finish.
@@ -601,8 +608,10 @@ def __report_a_failed_simulation(self, sim_idx, inputs_json, mutex, error_event)
         """
         details = traceback.format_exc()
         where = "worker startup" if sim_idx is None else f"iteration {sim_idx}"
+        announced = False
         with suppress(Exception):
             error_event.set()
+            announced = True
 
         mutex.acquire()
         try:
@@ -614,6 +623,7 @@ def __report_a_failed_simulation(self, sim_idx, inputs_json, mutex, error_event)
                 _SimMonitor.reprint(f"Error on {where}:\n{details}")
         finally:
             mutex.release()
+        return announced
 
     def __run_single_simulation(self):
         """Runs a single simulation and returns the inputs and outputs.
@@ -1790,8 +1800,7 @@ def export_errors_to_json(self, filename):
         self._write_log_to_json(self.errors_log, filename)
 
 
-# Short enough that a dead worker is noticed promptly, long enough that the
-# polling costs nothing over a run that takes hours.
+# Prompt enough to notice a dead worker, cheap enough over a run of hours.
 _JOIN_POLL_SECONDS = 0.2
 _SHUTDOWN_GRACE_SECONDS = 5.0
 
@@ -1801,36 +1810,40 @@ def _ended_badly(worker):
     return worker.exitcode not in (None, 0)
 
 
-def _the_run_is_already_lost(processes, error_event):
-    """Whether anything says the run cannot finish.
+def _wait_for_the_workers(processes, seconds):
+    """Join every worker against one shared deadline, not one deadline each.
 
-    A worker that fails the ordinary way reports through the event and returns,
-    so it exits cleanly and its exit code says nothing. Waiting only on exit
-    codes leaves the parent sitting behind a sibling that is stuck.
+    ``monotonic`` rather than the wall clock, which a correction can move
+    underneath a shutdown already in progress.
     """
-    if any(_ended_badly(worker) for worker in processes):
-        return True
-    with suppress(Exception):
-        return bool(error_event.is_set())
-    return False
+    deadline = monotonic() + seconds
+    for worker in processes:
+        worker.join(timeout=max(0.0, deadline - monotonic()))
 
 
 def _stop_the_workers_still_running(processes, error_event, grace_period):
-    """Ask the rest to stop, then end the ones that cannot.
+    """Ask the rest to stop, end what cannot, kill what outlives that.
 
     Asked first because a worker between simulations reads the event and leaves
     with its logs intact. One blocked on a lock its dead sibling was holding
-    never reaches that check, and only ending it frees the run.
+    never reaches that check. Terminate runs no handlers, so it comes second,
+    and a worker can still ignore it.
     """
     with suppress(Exception):
         error_event.set()
-    deadline = time() + grace_period
+    _wait_for_the_workers(processes, grace_period)
+
     for worker in processes:
-        worker.join(timeout=max(0.0, deadline - time()))
+        if worker.is_alive():
+            with suppress(Exception):
+                worker.terminate()
+    _wait_for_the_workers(processes, grace_period)
+
     for worker in processes:
         if worker.is_alive():
-            worker.terminate()
-            worker.join(timeout=grace_period)
+            with suppress(Exception):
+                worker.kill()
+    _wait_for_the_workers(processes, grace_period)
 
 
 def _join_the_workers(processes, error_event, grace_period=_SHUTDOWN_GRACE_SECONDS):
@@ -1839,12 +1852,15 @@ def _join_the_workers(processes, error_event, grace_period=_SHUTDOWN_GRACE_SECON
     The lock the workers share belongs to the manager and is not released when
     its holder is killed, so a sibling can block on a lock nobody owns while an
     unbounded join waits with it. Nothing here bounds a run that is merely
-    slow: only an exit code or a reported failure says a worker has given up.
+    slow, and a reported failure does not either: a worker that reports has
+    left nothing behind, and its siblings stop of their own accord once they
+    finish the simulation in hand. Only a worker that died can hold the lock
+    for good.
     """
     while any(worker.is_alive() for worker in processes):
         for worker in processes:
             worker.join(timeout=_JOIN_POLL_SECONDS)
-        if _the_run_is_already_lost(processes, error_event):
+        if any(_ended_badly(worker) for worker in processes):
             _stop_the_workers_still_running(processes, error_event, grace_period)
             return
 
@@ -1852,8 +1868,7 @@ def _join_the_workers(processes, error_event, grace_period=_SHUTDOWN_GRACE_SECON
 def _worker_failure_record(where, details):
     """A row for a worker that failed before it drew anything.
 
-    Written because the caller is told to read the error file, and a traceback
-    a worker printed is not there to be read once its output is redirected.
+    The caller is sent to this file, and a printed traceback is not in it.
     """
     return json.dumps({"index": None, "stage": where, "error": details}) + "\n"
 
@@ -1872,13 +1887,18 @@ def _indices_a_log_holds(path):
     return found
 
 
-def _refuse_a_run_that_lost_a_simulation(input_file, output_file, target):
+def _refuse_logs_missing_a_simulation(input_file, output_file, target):
     """Raise unless both logs hold every simulation the run was asked for.
 
     An exit code says how a worker ended, never whether the index it had
     already claimed reached the logs, and the monitor counts claims rather than
     rows. A worker that leaves between the two is invisible to everything else
     here, so the logs themselves are what the run is judged on.
+
+    Rows numbered past the target are left alone: an append given a smaller
+    target than the checkpoint already holds is an append question, not a lost
+    simulation. Streamed rather than read through ``_read_log_file``, which
+    would hold every row of a long study in memory to look at one field.
     """
     wanted = set(range(target))
     for label, path in (("input", input_file), ("output", output_file)):
@@ -1894,29 +1914,19 @@ def _refuse_a_run_that_lost_a_simulation(input_file, output_file, target):
                 f"The run is incomplete: the {label} log records "
                 f"{len(found) - len(held)} simulation(s) more than once."
             )
-        if held != wanted:
-            missing = sorted(wanted - held)
-            extra = sorted(held - wanted)
-            trouble = []
-            if missing:
-                trouble.append(
-                    f"{len(missing)} of {target} are missing, the first "
-                    f"being {missing[0]}"
-                )
-            if extra:
-                trouble.append(f"{len(extra)} are numbered past the run")
+        missing = sorted(wanted - held)
+        if missing:
             raise RuntimeError(
-                f"The run is incomplete: the {label} log does not hold every "
-                f"simulation that was asked for, {' and '.join(trouble)}."
+                f"The run is incomplete: the {label} log is missing "
+                f"{len(missing)} of {target} simulations, the first being "
+                f"{missing[0]}."
             )
 
 
 def _refuse_a_worker_that_did_not_finish(processes):
     """Raise if any worker left without exiting cleanly.
 
-    The workers report their own failures through an event, which one that was
-    killed never reaches, so what is left of it is its exit code. A negative
-    one is the signal that ended it, and ``None`` is one still running.
+    A negative code is the signal that ended it, ``None`` one still running.
     """
     unfinished = [
         f"worker {position} with exit code {process.exitcode}"
diff --git a/tests/unit/simulation/test_monte_carlo_parallel_runs.py b/tests/unit/simulation/test_monte_carlo_parallel_runs.py
index 4ab0be440..21ac90589 100644
--- a/tests/unit/simulation/test_monte_carlo_parallel_runs.py
+++ b/tests/unit/simulation/test_monte_carlo_parallel_runs.py
@@ -7,6 +7,7 @@
 def test_a_monte_carlo_run_finishes(
     stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path, parallel
 ):
+    """A real run completes and records every simulation, both modes."""
     # The parallel path hands each worker a SeedSequence rather than an int, and
     # nothing else in the suite exercises that. A worker that dies on it is not
     # reported, so this reads as a hang rather than as a failure.
diff --git a/tests/unit/simulation/test_monte_carlo_run_completeness.py b/tests/unit/simulation/test_monte_carlo_run_completeness.py
index 005eb31d6..9500a52b2 100644
--- a/tests/unit/simulation/test_monte_carlo_run_completeness.py
+++ b/tests/unit/simulation/test_monte_carlo_run_completeness.py
@@ -8,7 +8,7 @@
 from rocketpy.simulation import monte_carlo as mc_module
 from rocketpy.simulation.monte_carlo import (
     MonteCarlo,
-    _refuse_a_run_that_lost_a_simulation,
+    _refuse_logs_missing_a_simulation,
 )
 
 
@@ -31,49 +31,57 @@ def _complete(tmp_path, count=3, name="ok"):
 
 
 def test_a_run_that_recorded_everything_is_accepted(tmp_path):
+    """Logs holding every index the run asked for raise nothing."""
     inputs, outputs = _complete(tmp_path)
 
-    _refuse_a_run_that_lost_a_simulation(inputs, outputs, 3)
+    _refuse_logs_missing_a_simulation(inputs, outputs, 3)
 
 
 def test_a_missing_simulation_is_refused(tmp_path):
+    """A gap in the output log names the first index that is missing."""
     inputs, outputs = _complete(tmp_path)
     _a_log(tmp_path, "ok.outputs.txt", [_row(0), _row(2)])
 
     with pytest.raises(RuntimeError, match=r"output log.*missing.*being 1"):
-        _refuse_a_run_that_lost_a_simulation(inputs, outputs, 3)
+        _refuse_logs_missing_a_simulation(inputs, outputs, 3)
 
 
 def test_a_simulation_recorded_twice_is_refused(tmp_path):
+    """A duplicated index is refused, since a set alone would hide it."""
     inputs, outputs = _complete(tmp_path)
     _a_log(tmp_path, "ok.inputs.txt", [_row(0), _row(1), _row(1), _row(2)])
 
     with pytest.raises(RuntimeError, match="more than once"):
-        _refuse_a_run_that_lost_a_simulation(inputs, outputs, 3)
+        _refuse_logs_missing_a_simulation(inputs, outputs, 3)
 
 
 def test_a_row_that_cannot_be_read_is_refused(tmp_path):
+    """A torn row means the log's contents cannot be established."""
     inputs, outputs = _complete(tmp_path)
     _a_log(tmp_path, "ok.outputs.txt", [_row(0), "{half a row\n", _row(2)])
 
     with pytest.raises(RuntimeError, match="cannot be read"):
-        _refuse_a_run_that_lost_a_simulation(inputs, outputs, 3)
+        _refuse_logs_missing_a_simulation(inputs, outputs, 3)
 
 
-def test_a_row_numbered_past_the_run_is_refused(tmp_path):
-    inputs, outputs = _complete(tmp_path)
-    _a_log(tmp_path, "ok.inputs.txt", [_row(0), _row(1), _row(2), _row(9)])
+def test_rows_numbered_past_the_run_are_left_alone(tmp_path):
+    """An append below what a checkpoint already holds loses no simulation."""
+    # Refusing these said rows were missing when they were extra, and moved
+    # what append means inside a change about worker failure.
+    rows = [_row(index) for index in range(4)]
+    inputs = _a_log(tmp_path, "big.inputs.txt", rows)
+    outputs = _a_log(tmp_path, "big.outputs.txt", rows)
 
-    with pytest.raises(RuntimeError, match="past the run"):
-        _refuse_a_run_that_lost_a_simulation(inputs, outputs, 3)
+    _refuse_logs_missing_a_simulation(inputs, outputs, 2)
 
 
 def test_logs_that_hold_different_simulations_are_refused(tmp_path):
+    """The input and output logs have to hold the same indices."""
     inputs = _a_log(tmp_path, "a.inputs.txt", [_row(0), _row(1)])
     outputs = _a_log(tmp_path, "a.outputs.txt", [_row(0), _row(2)])
 
     with pytest.raises(RuntimeError):
-        _refuse_a_run_that_lost_a_simulation(inputs, outputs, 2)
+        _refuse_logs_missing_a_simulation(inputs, outputs, 2)
 
 
 def _leave_cleanly_without_recording(_flight):
@@ -85,6 +93,7 @@ def _leave_cleanly_without_recording(_flight):
 def test_a_worker_that_leaves_cleanly_without_recording_is_not_a_success(
     stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path
 ):
+    """A zero exit with no row written makes ``simulate`` raise."""
     analysis = MonteCarlo(
         filename=str(tmp_path / "study"),
         environment=stochastic_environment,
@@ -100,6 +109,7 @@ def test_a_worker_that_leaves_cleanly_without_recording_is_not_a_success(
 
 
 def test_no_failure_path_waits_on_a_worker_without_a_bound():
+    """No ``join`` in the parallel path is called without a timeout."""
     # An unbounded join anywhere in the parallel path puts back the hang that
     # the bounded teardown exists to end, and it does so where it is hardest
     # to notice: only when a worker is already stuck.
diff --git a/tests/unit/simulation/test_monte_carlo_worker_exit.py b/tests/unit/simulation/test_monte_carlo_worker_exit.py
index d9ca0dc68..5b44bd352 100644
--- a/tests/unit/simulation/test_monte_carlo_worker_exit.py
+++ b/tests/unit/simulation/test_monte_carlo_worker_exit.py
@@ -14,20 +14,24 @@ def _worker(exitcode):
 
 
 def test_workers_that_all_exited_cleanly_are_accepted():
+    """A fleet that all exited zero raises nothing."""
     _refuse_a_worker_that_did_not_finish([_worker(0), _worker(0)])
 
 
 def test_a_worker_killed_by_a_signal_is_refused():
+    """A negative exit code names the worker and the signal that ended it."""
     with pytest.raises(RuntimeError, match=r"worker 1 with exit code -9"):
         _refuse_a_worker_that_did_not_finish([_worker(0), _worker(-9)])
 
 
 def test_a_worker_that_exited_nonzero_is_refused():
+    """A positive exit code is refused the same way a signal is."""
     with pytest.raises(RuntimeError, match=r"worker 0 with exit code 1"):
         _refuse_a_worker_that_did_not_finish([_worker(1), _worker(0)])
 
 
 def test_every_unfinished_worker_is_named():
+    """The message names each unfinished worker and leaves the clean ones out."""
     with pytest.raises(RuntimeError) as raised:
         _refuse_a_worker_that_did_not_finish([_worker(-9), _worker(0), _worker(3)])
 
@@ -38,11 +42,12 @@ def test_every_unfinished_worker_is_named():
 
 @pytest.mark.parametrize("exitcode", [None, -15, 2])
 def test_anything_but_a_clean_exit_is_refused(exitcode):
+    """``None`` counts as unfinished, not as finished."""
     with pytest.raises(RuntimeError):
         _refuse_a_worker_that_did_not_finish([_worker(exitcode), _worker(0)])
 
 
-def _leave_without_recording(flight):  # pylint: disable=unused-argument
+def _leave_without_recording(_flight):
     """Ends the worker the way a kill or an out-of-memory exit does.
 
     ``os._exit`` rather than a signal, since ``SIGKILL`` is POSIX-only, and
@@ -55,6 +60,7 @@ def _leave_without_recording(flight):  # pylint: disable=unused-argument
 def test_a_worker_that_leaves_early_does_not_pass_as_a_finished_run(
     stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path
 ):
+    """A worker leaving through ``os._exit`` makes ``simulate`` raise."""
     # The event the workers report through is set by their own handler, and
     # this one leaves without running it, so the run used to return as though
     # it had done every simulation it was asked for.
diff --git a/tests/unit/simulation/test_monte_carlo_worker_join.py b/tests/unit/simulation/test_monte_carlo_worker_join.py
index 20ef8df22..5d93aea8f 100644
--- a/tests/unit/simulation/test_monte_carlo_worker_join.py
+++ b/tests/unit/simulation/test_monte_carlo_worker_join.py
@@ -10,19 +10,23 @@ class _Worker:
     which is the case an unbounded join waits out forever.
     """
 
-    def __init__(self, exitcode=0, alive_for=0, never=False):
+    def __init__(self, exitcode=0, alive_for=0, never=False, ignores_terminate=False):
         self.exitcode = None
         self._final_exitcode = exitcode
         self._alive_for = alive_for
         self._never = never
+        self._ignores_terminate = ignores_terminate
         self.joins = 0
+        self.timeouts = []
         self.terminated = False
+        self.killed = False
 
     def is_alive(self):
         return self.exitcode is None
 
-    def join(self, timeout=None):  # pylint: disable=unused-argument
+    def join(self, timeout=None):
         self.joins += 1
+        self.timeouts.append(timeout)
         # A real worker that never returns makes the caller hang, which is the
         # bug. Reproducing that here would hang CI instead of reporting, so the
         # stand-in gives up and says so.
@@ -33,7 +37,12 @@ def join(self, timeout=None):  # pylint: disable=unused-argument
 
     def terminate(self):
         self.terminated = True
-        self.exitcode = -15
+        if not self._ignores_terminate:
+            self.exitcode = -15
+
+    def kill(self):
+        self.killed = True
+        self.exitcode = -9
 
 
 class _Event:
@@ -47,14 +56,8 @@ def set(self):
         self.was_set = True
 
 
-class _BrokenEvent(_Event):
-    """A manager proxy that has gone away."""
-
-    def is_set(self):
-        raise OSError("the manager is gone")
-
-
 def test_a_run_where_every_worker_finishes_is_left_alone():
+    """A healthy fleet is joined to completion and never terminated."""
     workers = [_Worker(alive_for=3), _Worker(alive_for=5)]
 
     _join_the_workers(workers, _Event(), grace_period=0)
@@ -64,6 +67,7 @@ def test_a_run_where_every_worker_finishes_is_left_alone():
 
 
 def test_a_worker_blocked_behind_a_dead_one_does_not_wait_forever():
+    """One bad exit ends the wait for a sibling that never returns."""
     # The one that mattered. Without a bound this call never returns, so the
     # parent never reaches the check that would have reported the failure.
     died = _Worker(exitcode=-9, alive_for=1)
@@ -75,6 +79,7 @@ def test_a_worker_blocked_behind_a_dead_one_does_not_wait_forever():
 
 
 def test_the_survivors_are_asked_before_they_are_ended():
+    """The event is set before anything is terminated."""
     died = _Worker(exitcode=1, alive_for=1)
     blocked = _Worker(never=True)
     event = _Event()
@@ -85,6 +90,7 @@ def test_the_survivors_are_asked_before_they_are_ended():
 
 
 def test_a_survivor_that_stops_on_its_own_is_not_terminated():
+    """A worker that leaves during the grace period is left alone."""
     died = _Worker(exitcode=1, alive_for=1)
     cooperative = _Worker(alive_for=2)
 
@@ -96,6 +102,7 @@ def test_a_survivor_that_stops_on_its_own_is_not_terminated():
 
 @pytest.mark.parametrize("exitcode", [-9, 1, 2])
 def test_any_bad_exit_starts_the_shutdown(exitcode):
+    """Signals and non-zero codes both start the shutdown."""
     died = _Worker(exitcode=exitcode, alive_for=1)
     blocked = _Worker(never=True)
 
@@ -105,6 +112,7 @@ def test_any_bad_exit_starts_the_shutdown(exitcode):
 
 
 def test_a_slow_run_is_never_bounded():
+    """Elapsed time is not evidence: a slow fleet is polled, never stopped."""
     # Nothing here may act on how long a worker takes, only on it having died.
     slow = _Worker(alive_for=50)
     slower = _Worker(alive_for=80)
@@ -115,19 +123,23 @@ def test_a_slow_run_is_never_bounded():
     assert slower.joins > 50
 
 
-def test_a_reported_failure_also_stops_a_sibling_that_never_returns():
-    # A worker that fails the ordinary way is caught by the producer, reports
-    # through the event and returns, so it exits cleanly. Waiting only on exit
-    # codes leaves the parent sitting behind whichever sibling is stuck.
+def test_a_reported_failure_leaves_a_working_sibling_alone():
+    """A worker that only reported gives its siblings no reason to be ended."""
+    # Measured: with the event treated as a reason to stop, this sibling was
+    # terminated after three polls of the forty it needed. It was not stuck,
+    # it was mid-simulation, and it would have seen the event and left with
+    # its rows intact.
     reported = _Worker(exitcode=0, alive_for=1)
-    stuck = _Worker(never=True)
+    working = _Worker(alive_for=40)
 
-    _join_the_workers([reported, stuck], _Event(already_set=True), grace_period=0)
+    _join_the_workers([reported, working], _Event(already_set=True), grace_period=0)
 
-    assert stuck.terminated
+    assert not working.terminated
+    assert working.exitcode == 0
 
 
 def test_a_clean_run_is_not_stopped_by_an_event_nobody_set():
+    """An unset event leaves a healthy run running."""
     first, second = _Worker(alive_for=2), _Worker(alive_for=3)
 
     _join_the_workers([first, second], _Event(), grace_period=0)
@@ -135,11 +147,27 @@ def test_a_clean_run_is_not_stopped_by_an_event_nobody_set():
     assert not any(worker.terminated for worker in (first, second))
 
 
-def test_an_event_that_cannot_be_read_does_not_stop_the_run():
-    # The control on the control. If asking the manager raises, that is not
-    # evidence of failure and must not end a healthy run.
-    first, second = _Worker(alive_for=2), _Worker(alive_for=3)
+def test_a_worker_that_ignores_terminate_is_killed():
+    """Terminate can be ignored; the fleet still has to come down."""
+    died = _Worker(exitcode=-9, alive_for=1)
+    stubborn = _Worker(never=True, ignores_terminate=True)
 
-    _join_the_workers([first, second], _BrokenEvent(), grace_period=0)
+    _join_the_workers([died, stubborn], _Event(), grace_period=0)
 
-    assert not any(worker.terminated for worker in (first, second))
+    assert stubborn.terminated
+    assert stubborn.killed
+
+
+def test_the_fleet_comes_down_on_one_deadline_not_one_each():
+    """A stage gives the fleet one grace period between them, not each."""
+    # Observed through what each worker is offered: with a deadline of its own
+    # every worker is given the whole grace, so a fleet of thirty takes thirty
+    # times as long to give up on.
+    died = _Worker(exitcode=-9, alive_for=1)
+    stuck = [_Worker(never=True) for _ in range(4)]
+
+    _join_the_workers([died, *stuck], _Event(), grace_period=0.05)
+
+    offered = [t for t in stuck[-1].timeouts if t is not None]
+    assert offered
+    assert min(offered) < 0.05
diff --git a/tests/unit/simulation/test_monte_carlo_worker_reporting.py b/tests/unit/simulation/test_monte_carlo_worker_reporting.py
index cf9d51ebc..e437ac612 100644
--- a/tests/unit/simulation/test_monte_carlo_worker_reporting.py
+++ b/tests/unit/simulation/test_monte_carlo_worker_reporting.py
@@ -1,5 +1,6 @@
 import json
 import os
+from contextlib import suppress
 from types import SimpleNamespace
 
 import pytest
@@ -68,6 +69,7 @@ def _run(study, monitor, error_event, mutex=None):
 
 
 def test_a_worker_that_fails_before_seeding_finishes_says_so(tmp_path, capsys):
+    """A failure above the loop is reported against worker startup."""
     monitor = SimpleNamespace(keep_simulating=lambda: True)
     study, error_event = _a_worker(tmp_path, _refusing_model())
 
@@ -80,6 +82,8 @@ def test_a_worker_that_fails_before_seeding_finishes_says_so(tmp_path, capsys):
 
 
 def test_a_worker_that_fails_before_claiming_an_index_says_so(tmp_path, capsys):
+    """A failed claim is startup too, since no index was taken."""
+
     def refuse():
         raise RuntimeError("the monitor would not hand out an index")
 
@@ -96,6 +100,8 @@ def refuse():
 def test_a_worker_that_fails_inside_a_simulation_names_the_index(
     tmp_path, capsys, monkeypatch
 ):
+    """A failure after a claim is reported against that index."""
+
     # The control. An index is claimed and the simulation then fails, which is
     # the path that already worked, so the report still has to name it.
     def refuse(_self):
@@ -115,6 +121,7 @@ def refuse(_self):
 
 
 def test_a_startup_failure_is_written_down_and_not_only_printed(tmp_path):
+    """The error log gets a row even when no inputs were drawn."""
     # The caller is told to read the error file, and a traceback the worker
     # printed is not there to be read once its output has been redirected.
     monitor = SimpleNamespace(keep_simulating=lambda: True)
@@ -132,6 +139,8 @@ def test_a_startup_failure_is_written_down_and_not_only_printed(tmp_path):
 
 @pytest.mark.parametrize("failing", ["_set_stochastic", "increment"])
 def test_a_worker_failure_never_raises_out_of_the_producer(tmp_path, failing):
+    """A reported failure leaves the producer without an exception."""
+
     # The handler used to reach for names the loop had not bound yet, so the
     # process died with UnboundLocalError and the parent waited forever.
     def refuse(*_args):
@@ -154,6 +163,7 @@ def refuse(*_args):
 
 @pytest.mark.parametrize("breaking", ["error_file", "reprint", "event"])
 def test_reporting_a_failure_never_keeps_the_mutex(tmp_path, monkeypatch, breaking):
+    """The manager lock is released however the reporting goes."""
     # The mutex is the manager's, so a worker that ends while holding it leaves
     # the next one waiting on a process that is gone, and the parent never
     # reaches the join that would have noticed.
@@ -168,8 +178,13 @@ def test_reporting_a_failure_never_keeps_the_mutex(tmp_path, monkeypatch, breaki
     event = _ErrorEvent(refuse=breaking == "event")
     monitor = SimpleNamespace(keep_simulating=lambda: True)
     study, error_event = _a_worker(tmp_path, _refusing_model(), event)
+    mutex = _Mutex()
 
-    mutex = _run(study, monitor, error_event)
+    # A worker that could not announce its failure re-raises on the way out, so
+    # that its exit code carries what the event could not. The lock still has
+    # to be back either way, which is what this is about.
+    with suppress(RuntimeError):
+        _run(study, monitor, error_event, mutex)
 
     assert mutex.acquired == 1
     assert not mutex.held
@@ -178,6 +193,7 @@ def test_reporting_a_failure_never_keeps_the_mutex(tmp_path, monkeypatch, breaki
 def test_a_reporting_failure_does_not_replace_the_simulation_failure(
     tmp_path, monkeypatch, capsys
 ):
+    """An unwritable log does not hide what actually failed."""
     monkeypatch.setattr(mc_module, "_worker_failure_record", _raise_instead("no disk"))
     monitor = SimpleNamespace(keep_simulating=lambda: True)
     study, error_event = _a_worker(tmp_path, _refusing_model())
@@ -187,3 +203,90 @@ def test_a_reporting_failure_does_not_replace_the_simulation_failure(
     assert error_event.was_set
     assert "the models would not reseed" in capsys.readouterr().out
     assert not os.path.getsize(study.error_file)
+
+
+def _committing_producer(monkeypatch):
+    """Make one simulation run start to finish without a real flight."""
+    monkeypatch.setattr(
+        MonteCarlo, "_MonteCarlo__run_single_simulation", lambda self: None
+    )
+    monkeypatch.setattr(
+        MonteCarlo,
+        "_MonteCarlo__evaluate_flight_inputs",
+        lambda self, index: json.dumps({"index": index, "committed": True}) + "\n",
+    )
+    monkeypatch.setattr(
+        MonteCarlo,
+        "_MonteCarlo__evaluate_flight_outputs",
+        lambda self, flight, index: json.dumps({"index": index}) + "\n",
+    )
+
+
+def _one_then_broken():
+    calls = {"count": 0}
+
+    def keep_simulating():
+        calls["count"] += 1
+        if calls["count"] == 1:
+            return True
+        raise RuntimeError("the monitor died between simulations")
+
+    return SimpleNamespace(
+        keep_simulating=keep_simulating,
+        increment=lambda: 1,
+        print_update_status=lambda: None,
+    )
+
+
+def test_a_failure_between_simulations_is_not_blamed_on_the_last_one(
+    tmp_path, capsys, monkeypatch
+):
+    """A failure after a committed row is not reported against it."""
+    # Simulation 0 finishes and its row is committed. The next claim then
+    # fails, which is not simulation 0's doing and must not be recorded as it.
+    _committing_producer(monkeypatch)
+    model = SimpleNamespace(last_rnd_dict={}, _set_stochastic=lambda _seed: None)
+    study, error_event = _a_worker(tmp_path, model)
+
+    _run(study, _one_then_broken(), error_event)
+
+    assert "worker startup" in capsys.readouterr().out
+
+
+def test_a_committed_row_is_not_written_to_the_error_log_as_well(tmp_path, monkeypatch):
+    """A row that succeeded appears in one log, not in both."""
+    _committing_producer(monkeypatch)
+    model = SimpleNamespace(last_rnd_dict={}, _set_stochastic=lambda _seed: None)
+    study, error_event = _a_worker(tmp_path, model)
+
+    _run(study, _one_then_broken(), error_event)
+
+    with open(study.output_file, "r", encoding="utf-8") as written:
+        committed = [json.loads(line) for line in written if line.strip()]
+    with open(study.error_file, "r", encoding="utf-8") as recorded:
+        errored = [json.loads(line) for line in recorded if line.strip()]
+
+    assert committed == [{"index": 0}]
+    assert all(row.get("committed") is None for row in errored)
+
+
+def test_a_worker_that_cannot_announce_its_failure_does_not_exit_cleanly(tmp_path):
+    """With the event unreachable the producer raises, so the exit is not zero."""
+    # The event is how a worker reaches the parent. With it unreachable, the
+    # only signal left is how the process ends, so it must not end well.
+    model = _refusing_model()
+    study, error_event = _a_worker(tmp_path, model, _ErrorEvent(refuse=True))
+
+    with pytest.raises(RuntimeError, match="the models would not reseed"):
+        _run(study, SimpleNamespace(keep_simulating=lambda: True), error_event)
+
+
+def test_a_worker_that_did_announce_its_failure_returns(tmp_path):
+    """With the event delivered the producer returns on purpose."""
+    # The control. With the event delivered the parent already knows, so the
+    # producer returns and the process exits cleanly on purpose.
+    study, error_event = _a_worker(tmp_path, _refusing_model())
+
+    _run(study, SimpleNamespace(keep_simulating=lambda: True), error_event)
+
+    assert error_event.was_set

From eee7446ac8fa7962b4fd9d252bc56b055d0c777e Mon Sep 17 00:00:00 2001
From: thc1006 <84045975+thc1006@users.noreply.github.com>
Date: Tue, 25 Aug 2026 23:28:13 +0800
Subject: [PATCH 08/10] TST: read a log that an interrupted write left blank
 lines in

Codecov found the branch: a blank line is skipped rather than counted
as a row that cannot be read, and nothing exercised it. An interrupted
write is exactly what leaves them, so reporting a damaged log there
would fail a run that lost nothing.

Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com>
---
 .../simulation/test_monte_carlo_run_completeness.py   | 11 +++++++++++
 1 file changed, 11 insertions(+)

diff --git a/tests/unit/simulation/test_monte_carlo_run_completeness.py b/tests/unit/simulation/test_monte_carlo_run_completeness.py
index 9500a52b2..c02de2164 100644
--- a/tests/unit/simulation/test_monte_carlo_run_completeness.py
+++ b/tests/unit/simulation/test_monte_carlo_run_completeness.py
@@ -37,6 +37,17 @@ def test_a_run_that_recorded_everything_is_accepted(tmp_path):
     _refuse_logs_missing_a_simulation(inputs, outputs, 3)
 
 
+def test_blank_lines_between_rows_are_not_simulations(tmp_path):
+    """A blank line is skipped rather than counted as an unreadable row."""
+    # An interrupted write leaves them, and reading one as a row would report
+    # a damaged log for a run that lost nothing.
+    rows = [_row(0), "\n", _row(1), "   \n", _row(2)]
+    inputs = _a_log(tmp_path, "gappy.inputs.txt", rows)
+    outputs = _a_log(tmp_path, "gappy.outputs.txt", rows)
+
+    _refuse_logs_missing_a_simulation(inputs, outputs, 3)
+
+
 def test_a_missing_simulation_is_refused(tmp_path):
     """A gap in the output log names the first index that is missing."""
     inputs, outputs = _complete(tmp_path)

From 141503a6319076f40ca3ab3b8057edaf11b0dcd6 Mon Sep 17 00:00:00 2001
From: thc1006 <84045975+thc1006@users.noreply.github.com>
Date: Tue, 25 Aug 2026 23:35:42 +0800
Subject: [PATCH 09/10] BUG: keep the simulation index out of a data
 collector's reach

A collector's values are merged over the output record after the index
has been written into it, and the merge lets the right-hand side win,
so a key called index replaced the number of the simulation the row
belonged to. Measured on a two-simulation run with a collector
returning 999: both rows were written as 999.

Every check on a finished run reads that field, num_of_loaded_sims
counts on it, and an import keys results by it, so a run with such a
collector was writing a log nothing downstream could read correctly.
It was quiet until the completeness check turned it into a refusal
that named the wrong cause.

Refused where the collector is handed over, alongside the existing
export_list clash, so it costs nothing and happens before a file is
opened. A collector under any other name is untouched, which has a
test of its own.

Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com>
---
 rocketpy/simulation/monte_carlo.py            | 10 +++++
 .../test_monte_carlo_run_completeness.py      | 41 +++++++++++++++++++
 2 files changed, 51 insertions(+)

diff --git a/rocketpy/simulation/monte_carlo.py b/rocketpy/simulation/monte_carlo.py
index 50f95b23b..7d62cc6dc 100644
--- a/rocketpy/simulation/monte_carlo.py
+++ b/rocketpy/simulation/monte_carlo.py
@@ -44,6 +44,9 @@
 # this is the only format it can both resume from and overwrite safely.
 _SIMULATION_LOG_SUFFIX = ".txt"
 
+# Which simulation a row belongs to. Every check on a finished run reads it.
+_SIMULATION_INDEX_KEY = "index"
+
 
 def _refuse_logs_this_run_cannot_write(
     input_file, output_file, error_file, export_config=None
@@ -1028,6 +1031,13 @@ def _check_data_collector(self, data_collector):
                         "Invalid 'data_collector' key! "
                         f"Variable names overwrites 'export_list' key '{key}'."
                     )
+                if key == _SIMULATION_INDEX_KEY:
+                    raise ValueError(
+                        f"Invalid 'data_collector' key '{key}'! It is the "
+                        f"number of the simulation the row belongs to, which "
+                        f"is written after the collectors run and cannot be "
+                        f"replaced by one."
+                    )
                 if not callable(callback):
                     raise ValueError(
                         f"Invalid value in 'data_collector' for key '{key}'! "
diff --git a/tests/unit/simulation/test_monte_carlo_run_completeness.py b/tests/unit/simulation/test_monte_carlo_run_completeness.py
index c02de2164..181907423 100644
--- a/tests/unit/simulation/test_monte_carlo_run_completeness.py
+++ b/tests/unit/simulation/test_monte_carlo_run_completeness.py
@@ -142,3 +142,44 @@ def test_no_failure_path_waits_on_a_worker_without_a_bound():
     ]
 
     assert not unbounded, f"join() with no timeout at lines {unbounded}"
+
+
+def test_a_collector_cannot_take_over_the_simulation_index(
+    stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path
+):
+    """A collector key called index is refused before the run touches a file."""
+    # Measured before this was refused: every row was written with the
+    # collector's value, so the log said 999 twice for a two-simulation run
+    # and every check that reads an index was reading the wrong thing.
+    # Refused when the collector is handed over, which is before any file
+    # is opened, rather than at the end of a run that is already spoilt.
+    with pytest.raises(ValueError, match="index"):
+        MonteCarlo(
+            filename=str(tmp_path / "study"),
+            environment=stochastic_environment,
+            rocket=stochastic_calisto,
+            flight=stochastic_flight,
+            data_collector={"index": lambda flight: 999},
+        )
+
+
+def test_a_collector_key_of_its_own_is_still_welcome(
+    stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path
+):
+    """The control. Only the one reserved name is refused."""
+    analysis = MonteCarlo(
+        filename=str(tmp_path / "ok"),
+        environment=stochastic_environment,
+        rocket=stochastic_calisto,
+        flight=stochastic_flight,
+        data_collector={"apogee_twice": lambda flight: 2 * flight.apogee},
+    )
+
+    analysis.simulate(number_of_simulations=1, append=False)
+
+    with open(analysis.output_file, "r", encoding="utf-8") as written:
+        row = json.loads(next(line for line in written if line.strip()))
+    # Not the value of the index: how a run numbers its simulations is
+    # settled elsewhere, and pinning it here would tie this to that.
+    assert "index" in row
+    assert "apogee_twice" in row

From 629cd878e8f2be3304ea0a5c916b0b477e8e5d55 Mon Sep 17 00:00:00 2001
From: thc1006 <84045975+thc1006@users.noreply.github.com>
Date: Tue, 25 Aug 2026 23:58:18 +0800
Subject: [PATCH 10/10] MNT: narrow what the failure path swallows, and log the
 change

Review practice on this repository is consistent about breadth: broad
exception handlers get asked about, and a guard that cannot fire gets
asked to go. Both applied here.

terminate and kill were wrapped against a process that dies between the
liveness check and the signal. multiprocess already swallows
ProcessLookupError there and retries once on OSError, so the wrapper
could not fire and is gone.

The three that remain say what they expect. A manager that has gone
away answers a proxy call with a broken connection or an EOF, a log
that cannot be written raises OSError, and a closed stream raises
OSError or ValueError. Anything else now reaches the caller.

One CHANGELOG line, since one pull request is one line here and the
workflow that would have written it has not run since #1112 (#1173).

Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com>
---
 CHANGELOG.md                       |  1 +
 rocketpy/simulation/monte_carlo.py | 42 +++++++++++++-----------------
 2 files changed, 19 insertions(+), 24 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 28f601312..d04177951 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -60,6 +60,7 @@ Attention: The newest changes should be on top -->
 
 ### Fixed
 
+- BUG: Report a Monte Carlo worker that fails instead of hanging or passing for a finished run [#1182](https://github.com/RocketPy-Team/RocketPy/pull/1182)
 - BUG: Sample `StochasticFlight` inputs once per simulation [#1126](https://github.com/RocketPy-Team/RocketPy/pull/1126) [#1090](https://github.com/RocketPy-Team/RocketPy/issues/1090)
 - BUG: Fix spurious `ValueError` from floating-point roundoff at exact tank depletion [#1166](https://github.com/RocketPy-Team/RocketPy/pull/1166)
 - BUG: Draw each declared eccentricity once per simulation [#1168](https://github.com/RocketPy-Team/RocketPy/pull/1168)
diff --git a/rocketpy/simulation/monte_carlo.py b/rocketpy/simulation/monte_carlo.py
index 7d62cc6dc..aa90ae8c6 100644
--- a/rocketpy/simulation/monte_carlo.py
+++ b/rocketpy/simulation/monte_carlo.py
@@ -47,6 +47,9 @@
 # Which simulation a row belongs to. Every check on a finished run reads it.
 _SIMULATION_INDEX_KEY = "index"
 
+# How a manager that has gone away answers a proxy call.
+_MANAGER_IS_GONE = (OSError, EOFError)
+
 
 def _refuse_logs_this_run_cannot_write(
     input_file, output_file, error_file, export_config=None
@@ -612,16 +615,16 @@ def __report_a_failed_simulation(self, sim_idx, inputs_json, mutex, error_event)
         details = traceback.format_exc()
         where = "worker startup" if sim_idx is None else f"iteration {sim_idx}"
         announced = False
-        with suppress(Exception):
+        with suppress(_MANAGER_IS_GONE):
             error_event.set()
             announced = True
 
         mutex.acquire()
         try:
-            with suppress(Exception):
+            with suppress(OSError):
                 with open(self.error_file, "a", encoding="utf-8") as f:
                     f.write(inputs_json or _worker_failure_record(where, details))
-            with suppress(Exception):
+            with suppress(OSError, ValueError):
                 # Must use print() to remain visible from a worker process.
                 _SimMonitor.reprint(f"Error on {where}:\n{details}")
         finally:
@@ -1821,10 +1824,9 @@ def _ended_badly(worker):
 
 
 def _wait_for_the_workers(processes, seconds):
-    """Join every worker against one shared deadline, not one deadline each.
+    """Join every worker against one shared deadline, not one each.
 
-    ``monotonic`` rather than the wall clock, which a correction can move
-    underneath a shutdown already in progress.
+    Monotonic, since a clock correction would move a wall-clock deadline.
     """
     deadline = monotonic() + seconds
     for worker in processes:
@@ -1839,33 +1841,28 @@ def _stop_the_workers_still_running(processes, error_event, grace_period):
     never reaches that check. Terminate runs no handlers, so it comes second,
     and a worker can still ignore it.
     """
-    with suppress(Exception):
+    with suppress(_MANAGER_IS_GONE):
         error_event.set()
     _wait_for_the_workers(processes, grace_period)
 
     for worker in processes:
         if worker.is_alive():
-            with suppress(Exception):
-                worker.terminate()
+            worker.terminate()
     _wait_for_the_workers(processes, grace_period)
 
     for worker in processes:
         if worker.is_alive():
-            with suppress(Exception):
-                worker.kill()
+            worker.kill()
     _wait_for_the_workers(processes, grace_period)
 
 
 def _join_the_workers(processes, error_event, grace_period=_SHUTDOWN_GRACE_SECONDS):
-    """Wait for the workers, and stop waiting once one of them has died badly.
-
-    The lock the workers share belongs to the manager and is not released when
-    its holder is killed, so a sibling can block on a lock nobody owns while an
-    unbounded join waits with it. Nothing here bounds a run that is merely
-    slow, and a reported failure does not either: a worker that reports has
-    left nothing behind, and its siblings stop of their own accord once they
-    finish the simulation in hand. Only a worker that died can hold the lock
-    for good.
+    """Wait for the workers, and stop once one of them has died badly.
+
+    The shared lock belongs to the manager and outlives a killed holder, so a
+    sibling can block on a lock nobody owns. Neither slowness nor a reported
+    failure ends the wait: a worker that reports leaves nothing behind, and
+    its siblings stop once they finish the simulation in hand.
     """
     while any(worker.is_alive() for worker in processes):
         for worker in processes:
@@ -1876,10 +1873,7 @@ def _join_the_workers(processes, error_event, grace_period=_SHUTDOWN_GRACE_SECON
 
 
 def _worker_failure_record(where, details):
-    """A row for a worker that failed before it drew anything.
-
-    The caller is sent to this file, and a printed traceback is not in it.
-    """
+    """A row for a worker that failed before it drew anything."""
     return json.dumps({"index": None, "stage": where, "error": details}) + "\n"