@@ -14,6 +14,7 @@
 """
 
 import csv
+import hashlib
 import json
 import os
 import traceback
@@ -96,8 +97,75 @@ def _root_written_into_a_row(root_state):
     }
 
 
-def _root_a_log_was_written_with(path):
-    """The root every row of a log agrees on, or ``None`` if it holds none.
+_ROOT_FIELDS = frozenset(("entropy", "spawn_key", "pool_size", "n_children_spawned"))
+
+
+def _root_digest(root):
+    """A short stable name for a root, for rows that only have to match one.
+
+    Carried by the output rows instead of the root itself, which would cost a
+    hundred bytes a row on a study the input log already records it for.
+    """
+    canonical = json.dumps(root, sort_keys=True, separators=(",", ":"))
+    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:16]
+
+
+def _a_whole_number(value):
+    """A non-negative int, and not a bool standing in for one."""
+    return isinstance(value, int) and not isinstance(value, bool) and value >= 0
+
+
+def _whole_numbers(value, may_be_empty=False):
+    """A list of non-negative ints, empty only where that is a valid one."""
+    return (
+        isinstance(value, list)
+        and (may_be_empty or len(value) > 0)
+        and all(_a_whole_number(part) for part in value)
+    )
+
+
+def _root_state_a_row_records(root, path):
+    """The four values a root is rebuilt from, refused unless all are usable.
+
+    Agreeing rows show the study is one study, not that what they agree on can
+    be resumed: ``SeedSequence(entropy=None)`` draws fresh entropy every time.
+    """
+    entropy = root.get("entropy") if isinstance(root, dict) else None
+    usable = (
+        isinstance(root, dict)
+        and set(root) == _ROOT_FIELDS
+        and (_a_whole_number(entropy) or _whole_numbers(entropy))
+        and _whole_numbers(root.get("spawn_key"), may_be_empty=True)
+        and _a_whole_number(root.get("pool_size"))
+        and _a_whole_number(root.get("n_children_spawned"))
+    )
+    if usable:
+        state = (
+            entropy,
+            tuple(root["spawn_key"]),
+            root["pool_size"],
+            root["n_children_spawned"],
+        )
+        try:
+            # Drawing one child is what proves the pool size numpy will take.
+            _seed_of_simulation(state, 0)
+            return state
+        except ValueError:
+            pass
+    raise ValueError(
+        f"cannot continue {path}: its rows record a root that no stream "
+        f"can be rebuilt from, so what they were drawn with is unknown."
+    )
+
+
+def _rows_a_log_holds(path):
+    """How many simulations a log records, blank lines aside."""
+    with open(path, "r", encoding="utf-8") as recorded:
+        return sum(1 for line in recorded if line.strip())
+
+
+def _what_the_rows_say_drew_them(path):
+    """What every row of a log agrees drew it, or ``None`` if it holds none.
 
     ``None`` means the log holds no rows, and nothing else does. Rows that
     carry no root are refused instead: they cannot be shown to be one study,
@@ -116,14 +184,14 @@ def _root_a_log_was_written_with(path):
                     f"cannot continue {path}: a row cannot be read, so what "
                     f"produced it cannot be established."
                 ) from error
-            if _SIMULATION_ROOT_KEY not in row:
+            root = row.get(_SIMULATION_ROOT_KEY) if isinstance(row, dict) else None
+            if root is None:
                 raise ValueError(
                     f"cannot continue {path}: a row does not say which root "
                     f"drew it, which is how a study written before this "
                     f"release looks. Start a new one rather than continuing "
                     f"one whose rows cannot be checked."
                 )
-            root = row[_SIMULATION_ROOT_KEY]
             if first is _NOTHING_READ_YET:
                 first = root
             elif root != first:
@@ -733,16 +801,27 @@ def __continue_the_root_the_rows_carry(self, random_seed):
         also means a fresh object can continue a study, which is the ordinary
         way of resuming one.
         """
-        recorded = _root_a_log_was_written_with(self.input_file)
+        recorded = _what_the_rows_say_drew_them(self.input_file)
+        stamped = _what_the_rows_say_drew_them(self.output_file)
+        named = None if recorded is None else _root_digest(recorded)
+        if named != stamped:
+            raise ValueError(
+                f"cannot append to {self.input_file}: it and {self.output_file} "
+                f"were not drawn from the same root, so they are two studies "
+                f"rather than the two halves of one."
+            )
         if recorded is None:
             return
-        if random_seed is None:
-            self.__root_state = (
-                recorded["entropy"],
-                tuple(recorded["spawn_key"]),
-                recorded["pool_size"],
-                recorded["n_children_spawned"],
+        held = _rows_a_log_holds(self.input_file)
+        if held != _rows_a_log_holds(self.output_file):
+            raise ValueError(
+                f"cannot append to {self.input_file}: it and {self.output_file} "
+                f"hold different numbers of rows, so where to carry on from "
+                f"cannot be established."
             )
+        state = _root_state_a_row_records(recorded, self.input_file)
+        if random_seed is None:
+            self.__root_state = state
             return
         if _root_written_into_a_row(self.__root_state) != recorded:
             raise ValueError(
@@ -1010,8 +1089,6 @@ def __evaluate_flight_outputs(self, flight, sim_idx):
             export_item: getattr(flight, export_item)
             for export_item in self.export_list
         }
-        outputs_dict["index"] = sim_idx
-
         if self.data_collector is not None:
             additional_exports = {}
             for key, callback in self.data_collector.items():
@@ -1023,6 +1100,12 @@ def __evaluate_flight_outputs(self, flight, sim_idx):
                     ) from e
             outputs_dict = outputs_dict | additional_exports
 
+        # After the collectors: these two say which run the row belongs to.
+        outputs_dict["index"] = sim_idx
+        outputs_dict[_SIMULATION_ROOT_KEY] = _root_digest(
+            _root_written_into_a_row(self.__root_state)
+        )
+
         return (
             json.dumps(outputs_dict, cls=RocketPyEncoder, **self._export_config) + "\n"
         )
 
@@ -2,7 +2,11 @@
 
 import pytest
 
-from rocketpy.simulation.monte_carlo import MonteCarlo, _SIMULATION_ROOT_KEY
+from rocketpy.simulation.monte_carlo import (
+    MonteCarlo,
+    _root_digest,
+    _SIMULATION_ROOT_KEY,
+)
 
 
 def _study(tmp_path, stem, models):
@@ -112,6 +116,76 @@ def test_a_log_whose_rows_carry_no_root_is_refused(models, tmp_path):
         analysis.simulate(4, append=True)
 
 
+def _rewrite(path, replacement):
+    """Put ``replacement`` under the root key of every row of one log."""
+    with open(path, "r", encoding="utf-8") as written:
+        rows = [json.loads(line) for line in written if line.strip()]
+    for row in rows:
+        row[_SIMULATION_ROOT_KEY] = replacement(row[_SIMULATION_ROOT_KEY])
+    with open(path, "w", encoding="utf-8") as rewritten:
+        for row in rows:
+            rewritten.write(json.dumps(row) + "\n")
+
+
+def _rewrite_roots(analysis, damage):
+    """Damage the recorded root, leaving the two logs still naming one study.
+
+    The output rows hold the root's digest, so they are restamped rather than
+    damaged: otherwise the two logs disagree and that is refused first.
+    """
+    with open(analysis.input_file, "r", encoding="utf-8") as written:
+        first = json.loads(next(line for line in written if line.strip()))
+    damaged = damage(first[_SIMULATION_ROOT_KEY])
+    _rewrite(analysis.input_file, lambda _root: damaged)
+    if isinstance(damaged, dict):
+        _rewrite(analysis.output_file, lambda _digest: _root_digest(damaged))
+
+
+def test_a_log_whose_root_is_null_is_refused(models, tmp_path):
+    """A null root is no more of a lineage than no root at all."""
+    # It read as an empty log, so an append started a second root behind the
+    # damaged rows, which is the case this check exists to prevent.
+    analysis = _study(tmp_path, "study", models)
+    analysis.simulate(2, append=False, random_seed=42)
+
+    _rewrite_roots(analysis, lambda root: None)
+
+    with pytest.raises(ValueError, match="does not say which root"):
+        analysis.simulate(4, append=True)
+
+
+@pytest.mark.parametrize(
+    "damage",
+    [
+        lambda root: {**root, "entropy": None},
+        lambda root: {**root, "pool_size": 0},
+        lambda root: {**root, "pool_size": 2},
+        lambda root: {**root, "n_children_spawned": -1},
+        lambda root: {**root, "spawn_key": [True]},
+        lambda root: {key: value for key, value in root.items() if key != "entropy"},
+        lambda root: {**root, "unexpected": 1},
+    ],
+    ids=[
+        "null entropy",
+        "no pool",
+        "pool numpy refuses",
+        "negative base",
+        "bool key",
+        "short",
+        "long",
+    ],
+)
+def test_a_root_that_no_stream_can_be_rebuilt_from_is_refused(models, tmp_path, damage):
+    """Rows agreeing on a root does not make it a root that can be resumed."""
+    analysis = _study(tmp_path, "study", models)
+    analysis.simulate(2, append=False, random_seed=42)
+
+    _rewrite_roots(analysis, damage)
+
+    with pytest.raises(ValueError, match="rebuilt from"):
+        analysis.simulate(4, append=True)
+
+
 def test_an_empty_log_is_not_a_log_that_cannot_be_checked(models, tmp_path):
     """The control. Nothing recorded yet is a fresh start, not a refusal."""
     analysis = _study(tmp_path, "study", models)
@@ -168,3 +242,50 @@ def test_a_row_that_cannot_be_read_is_refused(models, tmp_path):
 
     with pytest.raises(ValueError, match="cannot be read"):
         analysis.simulate(4, append=True)
+
+
+def test_an_output_log_from_another_study_is_refused(models, tmp_path):
+    """Matching indices do not make two files the two halves of one run."""
+    # Nothing in the indices tells them apart: both studies number their rows
+    # from zero, so the root is what says they belong together.
+    one = _study(tmp_path, "one", models)
+    one.simulate(2, append=False, random_seed=42)
+    other = _study(tmp_path, "other", models)
+    other.simulate(2, append=False, random_seed=7)
+    with open(other.output_file, "r", encoding="utf-8") as theirs:
+        stolen = theirs.read()
+    with open(one.output_file, "w", encoding="utf-8") as ours:
+        ours.write(stolen)
+
+    with pytest.raises(ValueError, match="same root"):
+        one.simulate(4, append=True)
+
+
+def test_a_checkpoint_whose_halves_disagree_is_refused(models, tmp_path):
+    """Where to carry on from is not established by one log alone."""
+    analysis = _study(tmp_path, "study", models)
+    analysis.simulate(2, append=False, random_seed=42)
+    with open(analysis.output_file, "r", encoding="utf-8") as written:
+        rows = [line for line in written if line.strip()]
+    with open(analysis.output_file, "w", encoding="utf-8") as trimmed:
+        trimmed.write(rows[0])
+
+    with pytest.raises(ValueError, match="different numbers of rows"):
+        analysis.simulate(4, append=True)
+
+
+def test_a_collector_cannot_take_over_the_root(models, tmp_path):
+    """The root binds the two logs, so nothing outside the run writes it."""
+    analysis = _study(tmp_path, "study", models)
+    analysis.data_collector = {_SIMULATION_ROOT_KEY: lambda _flight: "forged"}
+
+    analysis.simulate(2, append=False, random_seed=42)
+
+    with open(analysis.output_file, "r", encoding="utf-8") as written:
+        roots = {
+            json.dumps(json.loads(line)[_SIMULATION_ROOT_KEY], sort_keys=True)
+            for line in written
+            if line.strip()
+        }
+    assert roots != {'"forged"'}
+    assert len(roots) == 1