From 7b2217a5f528843f6fae80ade8f456c16d4dc1d7 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 1/6] 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 761e9655c175c84d497c2425fb93303a34853543 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 2/6] 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 5000d583cc9f8b7b991f362580ec54ecdb782046 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 3/6] 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 2adc9d42d0963c85c8b849084107fbcf38486538 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 4/6] 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 6fb180f4ea4fa78118d64e2f71dd9c941bea6f48 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 5/6] 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 f5b992c72d47576e5bb7cbde95a25e653fd35530 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 6/6] 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))