From aa4178c1e654ed83dfcc2a862c43b2265b12ac7c Mon Sep 17 00:00:00 2001
From: Dhruv Kumar Singh <singhdhruvk@gmail.com>
Date: Sat, 1 Aug 2026 03:55:18 +0530
Subject: [PATCH 1/7] State the four Raft safety claims and check them after
 every run

---
 examples/raft/tests/checks.py           |  91 +++++++++++++++++++++
 examples/raft/tests/harness.py          |   7 ++
 examples/raft/tests/test_checks.py      | 101 ++++++++++++++++++++++++
 examples/raft/tests/test_replication.py |   4 +
 4 files changed, 203 insertions(+)
 create mode 100644 examples/raft/tests/checks.py
 create mode 100644 examples/raft/tests/test_checks.py

diff --git a/examples/raft/tests/checks.py b/examples/raft/tests/checks.py
new file mode 100644
index 0000000..1996250
--- /dev/null
+++ b/examples/raft/tests/checks.py
@@ -0,0 +1,91 @@
+"""The four Raft safety claims every intact run must satisfy."""
+
+from __future__ import annotations
+
+from itertools import combinations
+
+from raft.node import Event
+from raft.storage import Entry
+
+
+class InvariantViolation(AssertionError):
+    def __init__(self, invariant: str, detail: str) -> None:
+        super().__init__(f"{invariant}: {detail}")
+        self.invariant = invariant
+
+
+def check_invariants(
+    logs: dict[str, tuple[Entry, ...]], events: list[Event]
+) -> None:
+    _election_safety(events)
+    # Order matters: leader-completeness keeps only the first apply seen at an
+    # index, which is the whole story only once state-machine safety holds.
+    _state_machine_safety(events)
+    _leader_completeness(events)
+    _log_matching(logs)
+
+
+def _election_safety(events: list[Event]) -> None:
+    leaders: dict[int, set[str]] = {}
+    for event in events:
+        if event[0] == "leader":
+            _, name, term, _ = event
+            leaders.setdefault(term, set()).add(name)
+    for term, names in leaders.items():
+        if len(names) > 1:
+            raise InvariantViolation(
+                "election-safety", f"term {term} elected {sorted(names)}"
+            )
+
+
+def _state_machine_safety(events: list[Event]) -> None:
+    applied: dict[int, tuple[int, str]] = {}
+    for event in events:
+        if event[0] == "apply":
+            _, name, index, term, command = event
+            first = applied.setdefault(index, (term, command))
+            if first != (term, command):
+                raise InvariantViolation(
+                    "state-machine-safety",
+                    f"index {index} applied as {first} and, at {name}, "
+                    f"as {(term, command)}",
+                )
+
+
+def _leader_completeness(events: list[Event]) -> None:
+    # An entry committed under a term-T leader must appear in the log of
+    # every leader of a term above T. The floor records T at the moment of
+    # the first apply: the committing leader's own election event always
+    # precedes its applies, so the highest leader term seen so far is its
+    # term. Leaders at or below the floor -- stale-term stragglers whose
+    # counted votes predate the commit -- are outside the paper's claim.
+    committed: dict[int, tuple[Entry, int]] = {}
+    top_term = 0
+    for event in events:
+        if event[0] == "leader":
+            _, name, term, log = event
+            for index, (entry, floor) in committed.items():
+                if term > floor and (index > len(log) or log[index - 1] != entry):
+                    raise InvariantViolation(
+                        "leader-completeness",
+                        f"term-{term} leader {name} lacks committed "
+                        f"entry {index}: {entry}",
+                    )
+            top_term = max(top_term, term)
+        elif event[0] == "apply":
+            _, _, index, term, command = event
+            committed.setdefault(index, (Entry(term, command), top_term))
+
+
+def _log_matching(logs: dict[str, tuple[Entry, ...]]) -> None:
+    for a, b in combinations(sorted(logs), 2):
+        log_a, log_b = logs[a], logs[b]
+        for index in range(min(len(log_a), len(log_b)), 0, -1):
+            if log_a[index - 1].term == log_b[index - 1].term:
+                if log_a[:index] != log_b[:index]:
+                    raise InvariantViolation(
+                        "log-matching",
+                        f"{a} and {b} agree on the term at index {index} "
+                        "but not on the prefix before it",
+                    )
+                break
diff --git a/examples/raft/tests/harness.py b/examples/raft/tests/harness.py
index 8882332..d9aafd5 100644
--- a/examples/raft/tests/harness.py
+++ b/examples/raft/tests/harness.py
@@ -12,6 +12,8 @@
 from raft.node import LEADER, PORT, Event, RaftNode, Safeguards
 from raft.storage import Entry, MemoryStorage
 
+from checks import check_invariants
+
 
 def sim_loop() -> SimLoop:
     loop = asyncio.get_running_loop()
@@ -157,3 +159,8 @@ async def settle(cluster: Cluster, *, timeout_s: float = 120.0) -> None:
             if applied and applied[0] and all(a == applied[0] for a in applied):
                 return
             await asyncio.sleep(0.2)
+
+
+def verify(cluster: Cluster) -> None:
+    """Hold the run's whole history against the four safety claims."""
+    check_invariants(cluster.logs(), cluster.events)
diff --git a/examples/raft/tests/test_checks.py b/examples/raft/tests/test_checks.py
new file mode 100644
index 0000000..57683be
--- /dev/null
+++ b/examples/raft/tests/test_checks.py
@@ -0,0 +1,101 @@
+"""The safety checkers themselves, on synthetic histories."""
+
+from __future__ import annotations
+
+import pytest
+
+from raft.storage import Entry
+
+from checks import InvariantViolation, check_invariants
+
+
+def test_two_leaders_in_one_term_is_flagged() -> None:
+    events = [("leader", "n1", 3, ()), ("leader", "n2", 3, ())]
+    with pytest.raises(InvariantViolation) as caught:
+        check_invariants({}, events)
+    assert caught.value.invariant == "election-safety"
+
+
+def test_reelection_across_terms_is_fine() -> None:
+    check_invariants({}, [("leader", "n1", 3, ()), ("leader", "n2", 4, ())])
+
+
+def test_conflicting_applies_at_one_index_are_flagged() -> None:
+    events = [("apply", "n1", 1, 1, "a"), ("apply", "n2", 1, 2, "b")]
+    with pytest.raises(InvariantViolation) as caught:
+        check_invariants({}, events)
+    assert caught.value.invariant == "state-machine-safety"
+
+
+def test_matching_applies_are_fine() -> None:
+    check_invariants({}, [("apply", "n1", 1, 1, "a"), ("apply", "n2", 1, 1, "a")])
+
+
+def test_a_leader_missing_a_committed_entry_is_flagged() -> None:
+    events = [("apply", "n1", 1, 1, "a"), ("leader", "n2", 2, ())]
+    with pytest.raises(InvariantViolation) as caught:
+        check_invariants({}, events)
+    assert caught.value.invariant == "leader-completeness"
+
+
+def test_a_leader_carrying_the_committed_prefix_is_fine() -> None:
+    events = [
+        ("apply", "n1", 1, 1, "a"),
+        ("leader", "n2", 2, (Entry(1, "a"), Entry(1, "b"))),
+    ]
+    check_invariants({}, events)
+
+
+def test_a_leader_event_before_any_apply_is_fine() -> None:
+    check_invariants({}, [("leader", "n1", 1, ()), ("apply", "n2", 1, 1, "a")])
+
+
+def test_a_stale_term_leader_after_a_commit_is_not_judged() -> None:
+    events = [
+        ("leader", "n1", 3, (Entry(1, "a"),)),
+        ("apply", "n1", 1, 1, "a"),
+        ("leader", "n2", 2, ()),  # straggler election below the commit floor
+    ]
+    check_invariants({}, events)
+
+
+def test_a_higher_term_leader_missing_the_entry_still_fires() -> None:
+    events = [
+        ("leader", "n1", 3, (Entry(1, "a"),)),
+        ("apply", "n1", 1, 1, "a"),
+        ("leader", "n2", 4, ()),
+    ]
+    with pytest.raises(InvariantViolation) as caught:
+        check_invariants({}, events)
+    assert caught.value.invariant == "leader-completeness"
+
+
+def test_shared_terms_with_different_prefixes_are_flagged() -> None:
+    logs: dict[str, tuple[Entry, ...]] = {
+        "n1": (Entry(1, "a"), Entry(2, "c")),
+        "n2": (Entry(1, "b"), Entry(2, "c")),
+    }
+    with pytest.raises(InvariantViolation) as caught:
+        check_invariants(logs, [])
+    assert caught.value.invariant == "log-matching"
+
+
+def test_diverged_tails_with_distinct_terms_are_fine() -> None:
+    logs: dict[str, tuple[Entry, ...]] = {
+        "n1": (Entry(1, "a"), Entry(2, "x")),
+        "n2": (Entry(1, "a"), Entry(3, "y")),
+    }
+    check_invariants(logs, [])
+
+
+def test_a_clean_history_passes() -> None:
+    logs: dict[str, tuple[Entry, ...]] = {
+        "n1": (Entry(1, "a"),),
+        "n2": (Entry(1, "a"),),
+    }
+    events = [
+        ("leader", "n1", 1, ()),
+        ("apply", "n1", 1, 1, "a"),
+        ("apply", "n2", 1, 1, "a"),
+    ]
+    check_invariants(logs, events)
diff --git a/examples/raft/tests/test_replication.py b/examples/raft/tests/test_replication.py
index 0175b4a..d493326 100644
--- a/examples/raft/tests/test_replication.py
+++ b/examples/raft/tests/test_replication.py
@@ -22,6 +22,7 @@ async def test_commands_apply_everywhere_in_order() -> None:
     logs = [member.node.applied for member in cluster.members.values()]
     assert all(log == logs[0] for log in logs)
     assert [entry.command for entry in logs[0]] == ["k0", "k1", "k2"]
+    harness.verify(cluster)
 
 
 @sim_test
@@ -40,6 +41,7 @@ async def test_a_lagging_follower_catches_up() -> None:
         cluster.members[behind].node.applied
         == cluster.members[leader].node.applied
     )
+    harness.verify(cluster)
 
 
 @sim_test
@@ -55,6 +57,7 @@ async def test_committed_entries_survive_rolling_restarts() -> None:
     await harness.settle(cluster)
     for member in cluster.members.values():
         assert any(entry.command == "durable" for entry in member.node.applied)
+    harness.verify(cluster)
 
 
 @sim_test
@@ -76,3 +79,4 @@ async def test_a_deposed_leaders_unshared_entries_vanish() -> None:
         assert "lost" not in commands
         assert "keep" in commands
         assert "win" in commands
+    harness.verify(cluster)

From b351b9e88ebd93edb77e1de86e8cbe7eb691b1bb Mon Sep 17 00:00:00 2001
From: Dhruv Kumar Singh <singhdhruvk@gmail.com>
Date: Sat, 1 Aug 2026 04:12:53 +0530
Subject: [PATCH 2/7] Sweep Raft elections and replication under seeded chaos

---
 examples/raft/tests/harness.py             | 19 +++++++++++++
 examples/raft/tests/test_chaos_campaign.py | 33 ++++++++++++++++++++++
 2 files changed, 52 insertions(+)
 create mode 100644 examples/raft/tests/test_chaos_campaign.py

diff --git a/examples/raft/tests/harness.py b/examples/raft/tests/harness.py
index d9aafd5..4ceb855 100644
--- a/examples/raft/tests/harness.py
+++ b/examples/raft/tests/harness.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+import random
 from dataclasses import dataclass
 from typing import Any
 
@@ -82,6 +83,24 @@ async def restart(cluster: Cluster, name: str) -> None:
     _boot(cluster, name, member.storage)
 
 
+async def chaos(cluster: Cluster, rng: random.Random) -> None:
+    """A seed-derived fault schedule: partition windows and process restarts.
+
+    Cut sizes never exceed half the cluster, so a quorum side always exists
+    and the driver -- which is never partitioned -- can keep proposing.
+    """
+    loop = sim_loop()
+    for _ in range(3):
+        await asyncio.sleep(rng.uniform(0.5, 2.0))
+        cut = rng.sample(cluster.names, rng.randint(1, (len(cluster.names) - 1) // 2))
+        rest = [name for name in cluster.names if name not in cut]
+        loop.net.partition(cut, rest)
+        await asyncio.sleep(rng.uniform(0.5, 3.0))
+        loop.net.heal()
+        if rng.random() < 0.5:
+            await restart(cluster, rng.choice(cluster.names))
+
+
 def leader_now(cluster: Cluster) -> str | None:
     """Whoever claims leadership in the cluster's highest term, if anyone."""
     top = max(member.node.term for member in cluster.members.values())
diff --git a/examples/raft/tests/test_chaos_campaign.py b/examples/raft/tests/test_chaos_campaign.py
new file mode 100644
index 0000000..5f20e6d
--- /dev/null
+++ b/examples/raft/tests/test_chaos_campaign.py
@@ -0,0 +1,33 @@
+"""Hundreds of seeds of chaos; the four safety claims must hold on every one."""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from simloop import sim, sim_test
+
+import harness
+
+
+@pytest.mark.slow
+@sim_test(seeds=300)
+async def test_chaos_campaign_holds_the_invariants() -> None:
+    rng = sim.random
+    loop = harness.sim_loop()
+    cluster = await harness.start_cluster(size=5)
+    loop.net.set_defaults(latency=(0.01, 0.05), drop=0.02, duplicate=0.02)
+    for i in range(2):
+        await harness.propose(cluster, f"before.{i}")
+    disorder = loop.create_task(harness.chaos(cluster, rng))
+    sent = 0
+    while not disorder.done():
+        await harness.propose(cluster, f"during.{sent}", timeout_s=120.0)
+        sent += 1
+        await asyncio.sleep(rng.uniform(0.3, 1.0))
+    await disorder
+    loop.net.heal()
+    await harness.propose(cluster, "after", timeout_s=120.0)
+    await harness.settle(cluster)
+    harness.verify(cluster)

From 00e1c4d5de9e4bf46571856634d2736dbd3256f5 Mon Sep 17 00:00:00 2001
From: Dhruv Kumar Singh <singhdhruvk@gmail.com>
Date: Sat, 1 Aug 2026 04:30:55 +0530
Subject: [PATCH 3/7] Prove each safeguard load-bearing by removing it

---
 examples/raft/tests/harness.py        |  40 +++++++
 examples/raft/tests/test_ablations.py | 159 ++++++++++++++++++++++++++
 2 files changed, 199 insertions(+)
 create mode 100644 examples/raft/tests/test_ablations.py

diff --git a/examples/raft/tests/harness.py b/examples/raft/tests/harness.py
index 4ceb855..5b496e8 100644
--- a/examples/raft/tests/harness.py
+++ b/examples/raft/tests/harness.py
@@ -180,6 +180,46 @@ async def settle(cluster: Cluster, *, timeout_s: float = 120.0) -> None:
             await asyncio.sleep(0.2)
 
 
+async def figure_eight(safeguards: Safeguards) -> Cluster:
+    """The paper's Figure 8: an old-term entry reaches a quorum much later.
+
+    With the commit gate on, that entry may only commit once an entry of
+    the sitting leader's own term commits above it; with the gate off, a
+    counting leader commits it directly -- and a rival with a later-term
+    log can still erase it.
+    """
+    loop = sim_loop()
+    cluster = await start_cluster(size=5, safeguards=safeguards)
+    s1 = await wait_for_leader(cluster)
+    await propose(cluster, "a")
+    others = [name for name in cluster.names if name != s1]
+    buddy = others[0]
+    # "b" lands on s1 and buddy only, then the pair is cut off.
+    loop.net.partition([s1, buddy], others[1:])
+    await wire.call(s1, PORT, {"op": "propose", "command": "b"}, timeout_s=1.0)
+    await asyncio.sleep(0.5)
+    # The majority elects a new leader; it takes "c" and is cut before
+    # replicating it anywhere (on the seeds where the race lands that way).
+    s5 = await wait_for_leader(cluster, timeout_s=30.0, settle_s=1.0)
+    await wire.call(s5, PORT, {"op": "propose", "command": "c"}, timeout_s=1.0)
+    loop.net.heal()
+    loop.net.partition([s5], [name for name in cluster.names if name != s5])
+    # s1's side can now retake the cluster and spread "b" to a quorum. The
+    # two followers s5 left behind keep timing out and campaigning, and each
+    # doomed run -- their logs are short of s1's, so the up-to-date check
+    # refuses them -- drags the term up before s1 can win one of its own. The
+    # window has to cover those rounds plus the catch-up.
+    await asyncio.sleep(5.0)
+    loop.net.heal()
+    loop.net.partition([s1], [name for name in cluster.names if name != s1])
+    # With s1 gone and s5 back, s5's later-term log can win and erase "b".
+    # A settle() here would only ever time out: the gate is exercisable only
+    # with the no-op off, and s1 -- still cut away -- never catches up.
+    await propose(cluster, "d", timeout_s=60.0)
+    await asyncio.sleep(2.0)
+    return cluster
+
+
 def verify(cluster: Cluster) -> None:
     """Hold the run's whole history against the four safety claims."""
     check_invariants(cluster.logs(), cluster.events)
diff --git a/examples/raft/tests/test_ablations.py b/examples/raft/tests/test_ablations.py
new file mode 100644
index 0000000..34f6549
--- /dev/null
+++ b/examples/raft/tests/test_ablations.py
@@ -0,0 +1,159 @@
+"""Ablations: switch one safeguard off and prove the explorer catches it."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Callable, Coroutine
+from typing import Any
+
+import pytest
+
+from simloop import SeedReport, explore
+
+from raft.node import Safeguards
+
+import harness
+from checks import InvariantViolation
+
+BUDGET = 300
+
+Scenario = Callable[[], Coroutine[Any, Any, None]]
+
+
+def _find(scenario: Scenario, budget: int = BUDGET) -> SeedReport:
+    report = explore(scenario, range(budget))
+    assert report is not None, "ablation went undetected across the seed budget"
+    return report
+
+
+def test_double_voting_elects_two_leaders_in_one_term() -> None:
+    async def scenario() -> None:
+        cluster = await harness.start_cluster(
+            safeguards=Safeguards(one_vote_per_term=False)
+        )
+        loop = harness.sim_loop()
+        leader = await harness.wait_for_leader(cluster)
+        rest = [name for name in cluster.names if name != leader]
+        for _ in range(2):
+            loop.net.partition([leader], rest)
+            await asyncio.sleep(1.5)
+            loop.net.heal()
+            await asyncio.sleep(0.5)
+            harness.verify(cluster)
+
+    report = _find(scenario)
+    assert isinstance(report.exception, InvariantViolation)
+    assert report.exception.invariant == "election-safety"
+
+
+def test_unchecked_logs_let_a_stale_follower_lead() -> None:
+    async def scenario() -> None:
+        cluster = await harness.start_cluster(
+            safeguards=Safeguards(check_log_up_to_date=False)
+        )
+        loop = harness.sim_loop()
+        leader = await harness.wait_for_leader(cluster)
+        rest = [name for name in cluster.names if name != leader]
+        behind = rest[0]
+        loop.net.partition([behind], [leader, rest[1]])
+        for i in range(3):
+            await harness.propose(cluster, f"k{i}")
+        loop.net.heal()
+        loop.net.partition([leader], [behind, rest[1]])
+        await asyncio.sleep(2.0)
+        harness.verify(cluster)
+
+    report = _find(scenario)
+    assert isinstance(report.exception, InvariantViolation)
+    assert report.exception.invariant in (
+        "leader-completeness", "state-machine-safety",
+    )
+
+
+def test_skipped_persistence_forgets_committed_entries() -> None:
+    async def scenario() -> None:
+        cluster = await harness.start_cluster(
+            safeguards=Safeguards(persist_before_reply=False)
+        )
+        await harness.wait_for_leader(cluster)
+        for i in range(2):
+            await harness.propose(cluster, f"k{i}")
+        for name in list(cluster.names):
+            await harness.restart(cluster, name)
+            await asyncio.sleep(0.5)
+        await asyncio.sleep(2.0)
+        harness.verify(cluster)
+
+    report = _find(scenario)
+    assert isinstance(report.exception, InvariantViolation)
+    assert report.exception.invariant in (
+        "leader-completeness", "election-safety", "state-machine-safety",
+    )
+
+
+def test_accepting_stale_terms_rewrites_history() -> None:
+    async def scenario() -> None:
+        cluster = await harness.start_cluster(
+            safeguards=Safeguards(reject_stale_term=False)
+        )
+        loop = harness.sim_loop()
+        first = await harness.wait_for_leader(cluster)
+        await harness.propose(cluster, "k0")
+        rest = [name for name in cluster.names if name != first]
+        loop.net.partition([first], rest)
+        second = await harness.wait_for_leader(cluster)
+        await harness.propose(cluster, "k1")
+        await harness.propose(cluster, "k2")
+        loop.net.heal()
+        loop.net.partition([second], [name for name in cluster.names if name != second])
+        await asyncio.sleep(3.0)
+        loop.net.heal()
+        await harness.propose(cluster, "k3", timeout_s=60.0)
+        await asyncio.sleep(1.0)
+        harness.verify(cluster)
+
+    report = _find(scenario)
+    # No invariant pinned, deliberately: a node that answers superseded terms
+    # takes both stale appends and stale vote grants, which puts all four
+    # claims genuinely in reach -- naming all four would say nothing.
+    assert isinstance(report.exception, InvariantViolation)
+
+
+def test_committing_old_terms_by_count_loses_writes() -> None:
+    async def scenario() -> None:
+        cluster = await harness.figure_eight(
+            Safeguards(commit_own_term_only=False, leader_noop=False)
+        )
+        harness.verify(cluster)
+
+    report = _find(scenario, budget=500)
+    assert isinstance(report.exception, InvariantViolation)
+    assert report.exception.invariant in (
+        "state-machine-safety", "leader-completeness",
+    )
+
+
+@pytest.mark.slow
+def test_the_commit_gate_alone_keeps_history() -> None:
+    async def scenario() -> None:
+        cluster = await harness.figure_eight(Safeguards(leader_noop=False))
+        harness.verify(cluster)
+
+    assert explore(scenario, range(150)) is None
+
+
+@pytest.mark.slow
+def test_the_vote_ledger_alone_keeps_elections_single() -> None:
+    async def scenario() -> None:
+        cluster = await harness.start_cluster()
+        loop = harness.sim_loop()
+        leader = await harness.wait_for_leader(cluster)
+        rest = [name for name in cluster.names if name != leader]
+        for _ in range(2):
+            loop.net.partition([leader], rest)
+            await asyncio.sleep(1.5)
+            loop.net.heal()
+            await asyncio.sleep(0.5)
+            harness.verify(cluster)
+
+    assert explore(scenario, range(150)) is None

From 586af681ab6c1a6aaa60f624b615fe71a1fb6afa Mon Sep 17 00:00:00 2001
From: Dhruv Kumar Singh <singhdhruvk@gmail.com>
Date: Sat, 1 Aug 2026 05:39:18 +0530
Subject: [PATCH 4/7] Cancel connection handlers in a fixed order

---
 examples/raft/raft/node.py            | 10 +++--
 examples/raft/tests/test_ablations.py | 64 +++++++++++++++++++++------
 2 files changed, 57 insertions(+), 17 deletions(-)

diff --git a/examples/raft/raft/node.py b/examples/raft/raft/node.py
index f140569..1c6e2d8 100644
--- a/examples/raft/raft/node.py
+++ b/examples/raft/raft/node.py
@@ -99,15 +99,19 @@ def log(self) -> tuple[Entry, ...]:
     # ------------------------------------------------------------------
 
     async def run(self) -> None:
-        handlers: set[asyncio.Task[None]] = set()
+        # An insertion-ordered dict rather than a set: the sweep below cancels
+        # these in iteration order, and a set of tasks iterates by id(), which
+        # moves between runs -- two handlers swapping places there is enough to
+        # give one seed two different traces.
+        handlers: dict[asyncio.Task[None], None] = {}
 
         async def connection(
             reader: asyncio.StreamReader, writer: asyncio.StreamWriter
         ) -> None:
             task = asyncio.current_task()
             assert task is not None
-            handlers.add(task)
-            task.add_done_callback(handlers.discard)
+            handlers[task] = None
+            task.add_done_callback(lambda done: handlers.pop(done, None))
             await self._connection(reader, writer)
 
         server = await asyncio.start_server(connection, "0.0.0.0", self._port)
diff --git a/examples/raft/tests/test_ablations.py b/examples/raft/tests/test_ablations.py
index 34f6549..4fd15d1 100644
--- a/examples/raft/tests/test_ablations.py
+++ b/examples/raft/tests/test_ablations.py
@@ -16,6 +16,7 @@
 from checks import InvariantViolation
 
 BUDGET = 300
+REPLAYS = 8  # re-runs of one found seed in the replay-stability guard
 
 Scenario = Callable[[], Coroutine[Any, Any, None]]
 
@@ -70,21 +71,28 @@ async def scenario() -> None:
     )
 
 
-def test_skipped_persistence_forgets_committed_entries() -> None:
-    async def scenario() -> None:
-        cluster = await harness.start_cluster(
-            safeguards=Safeguards(persist_before_reply=False)
-        )
-        await harness.wait_for_leader(cluster)
-        for i in range(2):
-            await harness.propose(cluster, f"k{i}")
-        for name in list(cluster.names):
-            await harness.restart(cluster, name)
-            await asyncio.sleep(0.5)
-        await asyncio.sleep(2.0)
-        harness.verify(cluster)
+async def skipped_persistence() -> None:
+    """Module level, because the replay guard below explores it too.
 
-    report = _find(scenario)
+    It is the one ablation that restarts every process, which makes it the
+    scenario most likely to catch a node that lets something outside the
+    seed decide what happens next.
+    """
+    cluster = await harness.start_cluster(
+        safeguards=Safeguards(persist_before_reply=False)
+    )
+    await harness.wait_for_leader(cluster)
+    for i in range(2):
+        await harness.propose(cluster, f"k{i}")
+    for name in list(cluster.names):
+        await harness.restart(cluster, name)
+        await asyncio.sleep(0.5)
+    await asyncio.sleep(2.0)
+    harness.verify(cluster)
+
+
+def test_skipped_persistence_forgets_committed_entries() -> None:
+    report = _find(skipped_persistence)
     assert isinstance(report.exception, InvariantViolation)
     assert report.exception.invariant in (
         "leader-completeness", "election-safety", "state-machine-safety",
@@ -133,6 +141,34 @@ async def scenario() -> None:
     )
 
 
+@pytest.mark.slow
+def test_a_found_seed_replays_byte_identically() -> None:
+    """A seed's whole run must be a function of the seed and nothing else.
+
+    Held here rather than in the simulation's own hardening suite because
+    what it guards is on this side of the boundary: a node that iterates a
+    set of tasks, or otherwise lets id() order pick between two things the
+    schedule can tell apart, replays differently every few runs while still
+    failing the same way -- so nothing but the trace hash notices.
+
+    One replay is not enough to see that. The version of this node that
+    kept its connection handlers in a set held its order for the first
+    replay and broke on the second, every time it was measured, because
+    the heap is in much the same shape each time a run starts. REPLAYS is
+    set well past that: the check costs a fraction of a second either way.
+    """
+    first = explore(skipped_persistence, range(BUDGET))
+    assert first is not None, "ablation went undetected across the seed budget"
+    for attempt in range(REPLAYS):
+        again = explore(skipped_persistence, [first.seed])
+        assert again is not None, f"seed {first.seed} passed on replay {attempt}"
+        assert again.seed == first.seed
+        assert again.trace_hash == first.trace_hash, (
+            f"seed {first.seed} replayed differently on attempt {attempt}: "
+            f"{first.trace_hash} then {again.trace_hash}"
+        )
+
+
 @pytest.mark.slow
 def test_the_commit_gate_alone_keeps_history() -> None:
     async def scenario() -> None:

From a1627d42cf696d1f1560f7467176ea19173e330d Mon Sep 17 00:00:00 2001
From: Dhruv Kumar Singh <singhdhruvk@gmail.com>
Date: Sat, 1 Aug 2026 05:39:18 +0530
Subject: [PATCH 5/7] Record what the Raft campaign proves

---
 CHANGELOG.md            |   6 ++
 README.md               |  13 +++-
 examples/raft/README.md | 140 ++++++++++++++++++++++++++++++++++++++++
 3 files changed, 158 insertions(+), 1 deletion(-)
 create mode 100644 examples/raft/README.md

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 24a5990..d17e820 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,12 @@
 
 ## 0.2.0 (unreleased)
 
+- A second flagship demo: `examples/raft/` is a teaching-sized Raft (leader
+  election + log replication, plain asyncio on streams) tested only under
+  simulation — four safety invariants checked over 50,000 chaos seeds, five
+  safeguard ablations each caught and replayed from a seed, and failing
+  schedules minimized toward FIFO — down to a single interesting step in the
+  sharpest case.
 - Campaign evidence at scale, regenerable via `benchmarks/campaign.py`:
   100,000 seeds of jobqueue chaos green in six minutes on a laptop, every
   ablation caught with its failure density recorded, and 20 sampled failing
diff --git a/README.md b/README.md
index 1eb4303..34f8d84 100644
--- a/README.md
+++ b/README.md
@@ -178,7 +178,7 @@ net.crash("node2")                              # no reset, just silence
   stdlib outside one, so code under test can use entropy and clocks
   without breaking replay.
 
-## Proving it: the jobqueue demo
+## Proving it: the demos
 
 `examples/jobqueue/` is a complete distributed system — an exactly-once
 job scheduler (leases, fencing tokens, idempotency keys, backoff, and
@@ -188,6 +188,17 @@ poison jobs, and shows that removing any load-bearing safeguard produces
 a violation the explorer finds and replays from a seed. The bug table
 lives in [examples/jobqueue/README.md](https://github.com/dhruvl/simloop/blob/main/examples/jobqueue/README.md).
 
+`examples/raft/` is the second proof: a teaching-sized Raft — leader
+election and log replication in plain asyncio on streams — swept under
+50,000 seeds of partitions, crashed-and-restarted processes, and message
+loss. Four safety invariants hold on every seed; remove any safeguard
+(the vote ledger, the log-freshness check, the commit gate,
+persistence-before-reply, stale-term rejection) and the explorer finds a
+seed-replayable violation, then minimizes the failing schedule to the
+steps that had to go a particular way — one step out of 3,514 recorded, in
+the sharpest case. The table lives in
+[examples/raft/README.md](https://github.com/dhruvl/simloop/blob/main/examples/raft/README.md).
+
 ## Performance
 
 Simulation is cheap: SimLoop schedules a task step in ~4.4 µs (trace
diff --git a/examples/raft/README.md b/examples/raft/README.md
new file mode 100644
index 0000000..7ceb348
--- /dev/null
+++ b/examples/raft/README.md
@@ -0,0 +1,140 @@
+# raft — leader election and log replication, proven by simloop
+
+A teaching-sized Raft: about 500 lines of plain asyncio (stdlib only, no
+simloop imports) covering leader election, log replication, persistence
+across process restarts, and an applied state machine. Its test suite runs
+entirely under [simloop](../../README.md) — seeded scheduling, virtual time,
+simulated partitions, process restarts, message loss and duplication — and
+every failure it can produce replays exactly from a seed.
+
+**A demo, not a library.** No log compaction and no snapshots, no membership
+changes, no client sessions. Submission is at-least-once and says so: a
+command a leader accepts before being deposed can be resubmitted and commit
+twice, under two indices, and the safety claims below deliberately do not
+mind. Storage is an in-memory stand-in that survives a process restart within
+a run, the way a disk survives a reboot.
+
+## The claim, stated honestly
+
+The paper's four safety properties hold under every schedule the explorer
+reaches. Liveness is not claimed: a partitioned minority makes no progress,
+and the suite waits in virtual time rather than asserting any bound.
+
+Each rule that carries the safety argument sits behind its own flag in
+`Safeguards`, so the tests can switch exactly one off and watch the explorer
+find the schedule it lets through: stale-term rejection, one vote per term
+(§5.2), the log-freshness check on votes (§5.4.1), persistence before reply,
+the own-term commit gate (§5.4.2), and the per-term no-op that lets a quiet
+term commit (§8).
+
+Time is the only failure detector. Under simloop a partition stalls silently
+and a restarted process sends no reset, so a peer learns of trouble the way it
+would in production: from a request that never comes back.
+
+## Invariants
+
+Checked after every simulated run (`tests/checks.py`) against an ordered
+record of what the nodes observably did — `("leader", name, term, log)` at
+each election, `("apply", name, index, term, command)` at each state-machine
+apply — plus the logs the run ends with:
+
+1. **election safety** — at most one leader per term
+2. **log matching** — if two logs hold the same term at an index, they hold
+   the same prefix before it
+3. **leader completeness** — an entry committed under a term-T leader is in
+   the log of every leader of a term above T
+4. **state-machine safety** — no index is ever applied as two different
+   entries
+
+## The numbers
+
+- Scenario suite: 11 seeded scenarios (elections, replication, RPC framing)
+  × 10 seeds each, alongside unit tests for the log, vote and persistence
+  rules — 66 tests in all, green.
+- Campaign: **50,000 seeds** of five-node chaos — three randomized partition
+  windows per seed, a process restart after about half of them, 2% message
+  drop and 2% duplication throughout, with a client proposing — invariants
+  held on every seed. 917.29s (15m17s) with `--simloop-jobs=8` on an M4
+  MacBook Air, about 54 seeds a second. The same scenario runs 300 seeds
+  sequentially in 27.97s and 2,000 seeds in 36.34s at `--simloop-jobs=8`.
+- Replay stability: each ablation's found-at seed re-explored 100 times on a
+  fresh loop — 5 seeds × 100 replays, one trace hash apiece, byte-identical
+  throughout. The suite keeps re-earning the property:
+  `test_a_found_seed_replays_byte_identically` (slow-marked) finds a failing
+  seed and holds its trace hash across eight fresh replays.
+- Ablations: remove any load-bearing safeguard and the explorer finds a
+  violating schedule within a few seeds.
+
+| # | Safeguard removed | Invariant violated | Found at seed | Seeds searched | Reproduce |
+|---|---|---|---|---|---|
+| 1 | Vote ledger off (`one_vote_per_term=False`) | election-safety | 2 | 3 | `uv run pytest examples/raft/tests/test_ablations.py::test_double_voting_elects_two_leaders_in_one_term` |
+| 2 | Log-freshness check off (`check_log_up_to_date=False`) | leader-completeness | 0 | 1 | `... ::test_unchecked_logs_let_a_stale_follower_lead` |
+| 3 | Persistence before reply off (`persist_before_reply=False`) | leader-completeness | 4 | 5 | `... ::test_skipped_persistence_forgets_committed_entries` |
+| 4 | Stale-term rejection off (`reject_stale_term=False`) | state-machine-safety | 0 | 1 | `... ::test_accepting_stale_terms_rewrites_history` |
+| 5 | Commit gate off (`commit_own_term_only=False`, no-op off both sides) | state-machine-safety | 3 | 4 | `... ::test_committing_old_terms_by_count_loses_writes` |
+
+Rows 1–4 searched a budget of 300 seeds, row 5 a budget of 500. All five are
+labeled ablations — detection demonstrations, not bugs that were ever
+shipped.
+
+Two safeguards are also shown to be load-bearing *on their own*, which is the
+other half of the argument: with the commit gate the only thing standing (the
+per-term no-op switched off on both sides), the paper's Figure 8 runs clean
+across 150 seeds; with the vote ledger the only thing standing, the
+double-election scenario runs clean across 150 seeds. Both proofs are marked
+`slow`.
+
+### Found during development
+
+The replay-stability check above is what caught the last bug. `RaftNode.run`
+tracked its in-flight connection handlers in a `set` and cancelled them in
+iteration order when the incarnation died — but a set of tasks iterates by
+`id()`, which moves between runs, so two handlers could swap places. Seed 4
+of the persistence ablation produced two distinct trace hashes across 100
+re-runs: the same violation every time, reached by two different routes.
+An insertion-ordered dict fixed it. Nothing about the simulation was wrong:
+the demo was letting `id()` order decide something the schedule could see,
+which is exactly what the third rule in
+[docs/design.md](../../docs/design.md) tells the loop's own structures never
+to do. The found-at seeds in the table are unchanged either side of the fix.
+
+## What a failure looks like
+
+Each ablation test asserts that the explorer catches its ablation, so the
+tests themselves pass; the report is what the explorer hands back on the way.
+It names the invariant, the seed, and the trace around the failure. Run the
+same scenario through `explore(scenario, range(300), shrink=True)` and it
+also walks the recorded schedule back toward plain FIFO order, keeping only
+the decisions that have to go a particular way for the failure to reproduce.
+Stale-term rejection off, seed 0:
+
+    schedule shrink (experimental): 3,514 steps recorded, 57 runs to minimize
+    minimized: FIFO except step 1
+      step 1  TaskStepMethWrapper
+
+3,514 recorded scheduling steps; exactly one of them had to go a specific
+way. `FIFO throughout` is an answer too, and the double-voting ablation gives
+it across 1,169 steps — nothing about the task order matters there, so that
+race lives in the fault timing, in where the partition falls relative to the
+election timeouts.
+
+Re-run any detection from the table:
+
+    uv run pytest examples/raft/tests/test_ablations.py::test_accepting_stale_terms_rewrites_history
+
+Shrinking is experimental and costs extra runs; the `--simloop-shrink` flag
+reaches `@sim_test` tests, while the ablations call `explore()` directly and
+take `shrink=True` as an argument.
+
+## Run it
+
+    uv run pytest examples/raft/tests -q            # fast suite: scenarios, units, ablations
+    uv run pytest examples/raft/tests -q -m slow    # chaos campaign + the two safe proofs
+
+Turn the campaign up and spread the seeds over cores:
+
+    uv run pytest examples/raft/tests/test_chaos_campaign.py -q -m slow --simloop-seeds=50000 --simloop-jobs=8
+
+Replay any campaign failure exactly:
+
+    uv run pytest 'examples/raft/tests/test_chaos_campaign.py::test_chaos_campaign_holds_the_invariants' -m slow --simloop-replay=0

From 0ad6deab91a903b07c1d835266204261eb85e45c Mon Sep 17 00:00:00 2001
From: Dhruv Kumar Singh <singhdhruvk@gmail.com>
Date: Sat, 1 Aug 2026 05:39:18 +0530
Subject: [PATCH 6/7] Sweep the Raft demo nightly alongside the jobqueue

---
 .github/workflows/campaign.yml | 15 +++++++++------
 1 file changed, 9 insertions(+), 6 deletions(-)

diff --git a/.github/workflows/campaign.yml b/.github/workflows/campaign.yml
index aba6fa0..e8c47a4 100644
--- a/.github/workflows/campaign.yml
+++ b/.github/workflows/campaign.yml
@@ -1,10 +1,12 @@
 name: Nightly campaign
 
-# A small standing seed sweep of the jobqueue demo, so the campaign numbers
-# published in benchmarks/README.md keep being re-earned after every change
-# rather than dating from one good afternoon. Deliberately modest: a few
-# thousand seeds fits the runner's two cores in minutes, and the 100k-seed
-# sweeps stay a local, documented exercise.
+# A small standing seed sweep of both demos, so the campaign numbers published
+# in benchmarks/README.md and examples/raft/README.md keep being re-earned
+# after every change rather than dating from one good afternoon. Deliberately
+# modest: a few thousand seeds of each fit the runner's two cores in minutes,
+# and the sweeps at full scale stay a local, documented exercise. The raft
+# figure is well above the 300 seeds per-PR CI already runs on every matrix
+# leg, so the nightly is actually buying new coverage.
 
 on:
   schedule:
@@ -14,7 +16,7 @@ on:
 jobs:
   campaign:
     runs-on: ubuntu-latest
-    timeout-minutes: 30
+    timeout-minutes: 45
     steps:
       - uses: actions/checkout@v7
       - uses: astral-sh/setup-uv@v8.3.2
@@ -24,3 +26,4 @@ jobs:
       - run: uv run python benchmarks/campaign.py green --seeds 2000 --jobs 2 --checkpoint campaign-nightly.json
       - run: uv run python benchmarks/campaign.py ablations --seeds 500 --jobs 2 --checkpoint campaign-nightly-ablations.json
       - run: uv run python benchmarks/campaign.py stability --checkpoint campaign-nightly-ablations.json --sample 6 --reruns 20
+      - run: uv run pytest examples/raft/tests/test_chaos_campaign.py -q -m slow --simloop-seeds=5000 --simloop-jobs=2

From 2a4e2dbf6c7798ab2d85d8421eda4b5375759d51 Mon Sep 17 00:00:00 2001
From: Dhruv Kumar Singh <singhdhruvk@gmail.com>
Date: Sat, 1 Aug 2026 05:55:29 +0530
Subject: [PATCH 7/7] Pin down the committing term and surface crashed nodes

---
 examples/raft/README.md               | 34 +++++++++++++++++++--------
 examples/raft/raft/node.py            | 13 +++++++---
 examples/raft/tests/checks.py         | 20 ++++++++--------
 examples/raft/tests/harness.py        |  8 +++++++
 examples/raft/tests/test_ablations.py |  1 +
 examples/raft/tests/test_checks.py    | 33 +++++++++++++++++++-------
 examples/raft/tests/test_election.py  |  4 ++++
 examples/raft/tests/test_log.py       |  4 ++--
 8 files changed, 83 insertions(+), 34 deletions(-)

diff --git a/examples/raft/README.md b/examples/raft/README.md
index 7ceb348..db6eae1 100644
--- a/examples/raft/README.md
+++ b/examples/raft/README.md
@@ -18,7 +18,7 @@ a run, the way a disk survives a reboot.
 
 The paper's four safety properties hold under every schedule the explorer
 reaches. Liveness is not claimed: a partitioned minority makes no progress,
-and the suite waits in virtual time rather than asserting any bound.
+and the suite waits in virtual time rather than asserting a wall-clock bound.
 
 Each rule that carries the safety argument sits behind its own flag in
 `Safeguards`, so the tests can switch exactly one off and watch the explorer
@@ -35,8 +35,11 @@ would in production: from a request that never comes back.
 
 Checked after every simulated run (`tests/checks.py`) against an ordered
 record of what the nodes observably did — `("leader", name, term, log)` at
-each election, `("apply", name, index, term, command)` at each state-machine
-apply — plus the logs the run ends with:
+each election, `("apply", name, index, term, command, node_term)` at each
+state-machine apply — plus the logs the run ends with. The applier's own
+term rides along because leader completeness needs the term an entry was
+committed under, and the first apply of an index always happens on the
+leader that committed it:
 
 1. **election safety** — at most one leader per term
 2. **log matching** — if two logs hold the same term at an index, they hold
@@ -50,18 +53,19 @@ apply — plus the logs the run ends with:
 
 - Scenario suite: 11 seeded scenarios (elections, replication, RPC framing)
   × 10 seeds each, alongside unit tests for the log, vote and persistence
-  rules — 66 tests in all, green.
+  rules — 63 fast tests plus the slow proofs, green.
 - Campaign: **50,000 seeds** of five-node chaos — three randomized partition
   windows per seed, a process restart after about half of them, 2% message
   drop and 2% duplication throughout, with a client proposing — invariants
   held on every seed. 917.29s (15m17s) with `--simloop-jobs=8` on an M4
   MacBook Air, about 54 seeds a second. The same scenario runs 300 seeds
   sequentially in 27.97s and 2,000 seeds in 36.34s at `--simloop-jobs=8`.
-- Replay stability: each ablation's found-at seed re-explored 100 times on a
-  fresh loop — 5 seeds × 100 replays, one trace hash apiece, byte-identical
-  throughout. The suite keeps re-earning the property:
-  `test_a_found_seed_replays_byte_identically` (slow-marked) finds a failing
-  seed and holds its trace hash across eight fresh replays.
+- Replay stability: a one-off local measurement re-explored each ablation's
+  found-at seed 100 times on a fresh loop — 5 seeds × 100 replays, one trace
+  hash apiece, byte-identical throughout. The standing check is
+  `test_a_found_seed_replays_byte_identically` (slow-marked, so it rides the
+  nightly): it locates a failing seed and holds its trace hash across eight
+  fresh replays.
 - Ablations: remove any load-bearing safeguard and the explorer finds a
   violating schedule within a few seeds.
 
@@ -77,6 +81,13 @@ Rows 1–4 searched a budget of 300 seeds, row 5 a budget of 500. All five are
 labeled ablations — detection demonstrations, not bugs that were ever
 shipped.
 
+"Invariant violated" records what the found seed actually produced, not the
+only thing that ablation can produce. Three of the five tests deliberately
+accept any genuine violation class: a node that answers superseded terms, for
+instance, takes both stale appends and stale vote grants, so naming one of the
+four claims would say less than letting the checker report which one broke
+first.
+
 Two safeguards are also shown to be load-bearing *on their own*, which is the
 other half of the argument: with the commit gate the only thing standing (the
 per-term no-op switched off on both sides), the paper's Figure 8 runs clean
@@ -129,7 +140,10 @@ take `shrink=True` as an argument.
 ## Run it
 
     uv run pytest examples/raft/tests -q            # fast suite: scenarios, units, ablations
-    uv run pytest examples/raft/tests -q -m slow    # chaos campaign + the two safe proofs
+    uv run pytest examples/raft/tests -q -m slow    # campaign, the two safe proofs, the replay guard
+
+Add `-s` to either and the ablations print the explorer's report — the failing
+seed and the trace around it.
 
 Turn the campaign up and spread the seeds over cores:
 
diff --git a/examples/raft/raft/node.py b/examples/raft/raft/node.py
index 1c6e2d8..d4640a8 100644
--- a/examples/raft/raft/node.py
+++ b/examples/raft/raft/node.py
@@ -27,7 +27,9 @@
 LEADER = "leader"
 
 # One observable fact, appended in order: ("leader", name, term, log_snapshot)
-# or ("apply", name, index, term, command). The safety checks read these.
+# or ("apply", name, index, term, command, node_term), where node_term is the
+# applier's current term -- on the first apply of an index that is the term
+# whose leader committed it. The safety checks read these.
 Event = tuple[Any, ...]
 
 
@@ -72,7 +74,7 @@ def __init__(
         self._safeguards = safeguards if safeguards is not None else Safeguards()
         self._events = events
         self._state = storage.load()
-        self._quorum = (len(peers) + 1) // 2 + 1
+        self._quorum = (len(self._peers) + 1) // 2 + 1
         self.role = FOLLOWER
         self.commit_index = 0
         self.applied: list[Entry] = []
@@ -400,6 +402,11 @@ def _apply(self) -> None:
             self._applied_index += 1
             if entry.command:
                 self._record(
-                    "apply", self._name, self._applied_index, entry.term, entry.command
+                    "apply",
+                    self._name,
+                    self._applied_index,
+                    entry.term,
+                    entry.command,
+                    self._state.term,
                 )
                 self.applied.append(entry)
diff --git a/examples/raft/tests/checks.py b/examples/raft/tests/checks.py
index 1996250..ffeddb7 100644
--- a/examples/raft/tests/checks.py
+++ b/examples/raft/tests/checks.py
@@ -42,7 +42,7 @@ def _state_machine_safety(events: list[Event]) -> None:
     applied: dict[int, tuple[int, str]] = {}
     for event in events:
         if event[0] == "apply":
-            _, name, index, term, command = event
+            _, name, index, term, command, _node_term = event
             first = applied.setdefault(index, (term, command))
             if first != (term, command):
                 raise InvariantViolation(
@@ -54,13 +54,14 @@ def _state_machine_safety(events: list[Event]) -> None:
 
 def _leader_completeness(events: list[Event]) -> None:
     # An entry committed under a term-T leader must appear in the log of
-    # every leader of a term above T. The floor records T at the moment of
-    # the first apply: the committing leader's own election event always
-    # precedes its applies, so the highest leader term seen so far is its
-    # term. Leaders at or below the floor -- stale-term stragglers whose
-    # counted votes predate the commit -- are outside the paper's claim.
+    # every leader of a term above T. The floor is T itself, read straight
+    # off the first apply of the index: a follower only learns an index is
+    # committed from a later append carrying leader_commit, so the first
+    # apply of any index happens on the committing leader, and that event's
+    # node_term is the term it committed under. Leaders at or below the
+    # floor -- stale-term stragglers whose counted votes predate the commit
+    # -- are outside the paper's claim.
     committed: dict[int, tuple[Entry, int]] = {}
-    top_term = 0
     for event in events:
         if event[0] == "leader":
             _, name, term, log = event
@@ -71,10 +72,9 @@ def _leader_completeness(events: list[Event]) -> None:
                         f"term-{term} leader {name} lacks committed "
                         f"entry {index}: {entry}",
                     )
-            top_term = max(top_term, term)
         elif event[0] == "apply":
-            _, _, index, term, command = event
-            committed.setdefault(index, (Entry(term, command), top_term))
+            _, _, index, term, command, node_term = event
+            committed.setdefault(index, (Entry(term, command), node_term))
 
 
 def _log_matching(logs: dict[str, tuple[Entry, ...]]) -> None:
diff --git a/examples/raft/tests/harness.py b/examples/raft/tests/harness.py
index 5b496e8..6382fce 100644
--- a/examples/raft/tests/harness.py
+++ b/examples/raft/tests/harness.py
@@ -169,6 +169,14 @@ async def settle(cluster: Cluster, *, timeout_s: float = 120.0) -> None:
     """Wait (in virtual time) until every live member applied the same sequence."""
     async with asyncio.timeout(timeout_s):
         while True:
+            for member in cluster.members.values():
+                # A node that died of a real bug would otherwise just drop out
+                # of the quorum below and let the rest converge without it.
+                # Reading the result re-raises whatever killed it. Restarts
+                # cancel their task, and a cancelled task has no result to
+                # speak of, so those stay excluded quietly as before.
+                if member.task.done() and not member.task.cancelled():
+                    member.task.result()
             live = [
                 member.node
                 for member in cluster.members.values()
diff --git a/examples/raft/tests/test_ablations.py b/examples/raft/tests/test_ablations.py
index 4fd15d1..44c83c0 100644
--- a/examples/raft/tests/test_ablations.py
+++ b/examples/raft/tests/test_ablations.py
@@ -24,6 +24,7 @@
 def _find(scenario: Scenario, budget: int = BUDGET) -> SeedReport:
     report = explore(scenario, range(budget))
     assert report is not None, "ablation went undetected across the seed budget"
+    print(report.render())  # visible under `pytest -s`; the demo's best artifact
     return report
 
 
diff --git a/examples/raft/tests/test_checks.py b/examples/raft/tests/test_checks.py
index 57683be..0932882 100644
--- a/examples/raft/tests/test_checks.py
+++ b/examples/raft/tests/test_checks.py
@@ -21,18 +21,18 @@ def test_reelection_across_terms_is_fine() -> None:
 
 
 def test_conflicting_applies_at_one_index_are_flagged() -> None:
-    events = [("apply", "n1", 1, 1, "a"), ("apply", "n2", 1, 2, "b")]
+    events = [("apply", "n1", 1, 1, "a", 1), ("apply", "n2", 1, 2, "b", 2)]
     with pytest.raises(InvariantViolation) as caught:
         check_invariants({}, events)
     assert caught.value.invariant == "state-machine-safety"
 
 
 def test_matching_applies_are_fine() -> None:
-    check_invariants({}, [("apply", "n1", 1, 1, "a"), ("apply", "n2", 1, 1, "a")])
+    check_invariants({}, [("apply", "n1", 1, 1, "a", 1), ("apply", "n2", 1, 1, "a", 1)])
 
 
 def test_a_leader_missing_a_committed_entry_is_flagged() -> None:
-    events = [("apply", "n1", 1, 1, "a"), ("leader", "n2", 2, ())]
+    events = [("apply", "n1", 1, 1, "a", 1), ("leader", "n2", 2, ())]
     with pytest.raises(InvariantViolation) as caught:
         check_invariants({}, events)
     assert caught.value.invariant == "leader-completeness"
@@ -40,20 +40,21 @@ def test_a_leader_missing_a_committed_entry_is_flagged() -> None:
 
 def test_a_leader_carrying_the_committed_prefix_is_fine() -> None:
     events = [
-        ("apply", "n1", 1, 1, "a"),
+        ("apply", "n1", 1, 1, "a", 1),
         ("leader", "n2", 2, (Entry(1, "a"), Entry(1, "b"))),
     ]
     check_invariants({}, events)
 
 
 def test_a_leader_event_before_any_apply_is_fine() -> None:
-    check_invariants({}, [("leader", "n1", 1, ()), ("apply", "n2", 1, 1, "a")])
+    check_invariants({}, [("leader", "n1", 1, ()), ("apply", "n2", 1, 1, "a", 1)])
 
 
 def test_a_stale_term_leader_after_a_commit_is_not_judged() -> None:
     events = [
         ("leader", "n1", 3, (Entry(1, "a"),)),
-        ("apply", "n1", 1, 1, "a"),
+        # n1 sits in term 3 and commits an old term-1 entry: the floor is 3.
+        ("apply", "n1", 1, 1, "a", 3),
         ("leader", "n2", 2, ()),  # straggler election below the commit floor
     ]
     check_invariants({}, events)
@@ -62,7 +63,7 @@ def test_a_stale_term_leader_after_a_commit_is_not_judged() -> None:
 def test_a_higher_term_leader_missing_the_entry_still_fires() -> None:
     events = [
         ("leader", "n1", 3, (Entry(1, "a"),)),
-        ("apply", "n1", 1, 1, "a"),
+        ("apply", "n1", 1, 1, "a", 3),
         ("leader", "n2", 4, ()),
     ]
     with pytest.raises(InvariantViolation) as caught:
@@ -70,6 +71,20 @@ def test_a_higher_term_leader_missing_the_entry_still_fires() -> None:
     assert caught.value.invariant == "leader-completeness"
 
 
+def test_the_floor_is_the_committing_term_not_the_highest_seen() -> None:
+    # A term-9 election is on record before the commit, but the entry was
+    # committed under term 3 — a term-5 leader lacking it must still be
+    # judged, not exempted by the higher election that came first.
+    events = [
+        ("leader", "n9", 9, (Entry(1, "a"),)),
+        ("apply", "n1", 1, 1, "a", 3),
+        ("leader", "n2", 5, ()),
+    ]
+    with pytest.raises(InvariantViolation) as caught:
+        check_invariants({}, events)
+    assert caught.value.invariant == "leader-completeness"
+
+
 def test_shared_terms_with_different_prefixes_are_flagged() -> None:
     logs: dict[str, tuple[Entry, ...]] = {
         "n1": (Entry(1, "a"), Entry(2, "c")),
@@ -95,7 +110,7 @@ def test_a_clean_history_passes() -> None:
     }
     events = [
         ("leader", "n1", 1, ()),
-        ("apply", "n1", 1, 1, "a"),
-        ("apply", "n2", 1, 1, "a"),
+        ("apply", "n1", 1, 1, "a", 1),
+        ("apply", "n2", 1, 1, "a", 1),
     ]
     check_invariants(logs, events)
diff --git a/examples/raft/tests/test_election.py b/examples/raft/tests/test_election.py
index d77987c..fd4ee77 100644
--- a/examples/raft/tests/test_election.py
+++ b/examples/raft/tests/test_election.py
@@ -18,6 +18,7 @@ async def test_a_quiet_cluster_elects_exactly_one_leader() -> None:
     await asyncio.sleep(2.0)
     leaders = [m.name for m in cluster.members.values() if m.node.role == LEADER]
     assert leaders == [leader]
+    harness.verify(cluster)
 
 
 @sim_test
@@ -28,6 +29,7 @@ async def test_a_settled_leader_stays_leader() -> None:
     await asyncio.sleep(5.0)
     assert cluster.members[leader].node.role == LEADER
     assert cluster.members[leader].node.term == term
+    harness.verify(cluster)
 
 
 @sim_test
@@ -41,6 +43,7 @@ async def test_the_cluster_survives_a_leader_restart() -> None:
     assert cluster.members[leader].node.role == LEADER
     # The restart cut the heartbeats, so the survivors must have opened a term.
     assert cluster.members[leader].node.term > before
+    harness.verify(cluster)
 
 
 @sim_test
@@ -55,3 +58,4 @@ async def test_a_partitioned_leader_steps_down_on_heal() -> None:
     loop.net.heal()
     await asyncio.sleep(2.0)
     assert cluster.members[first].node.role != LEADER
+    harness.verify(cluster)
diff --git a/examples/raft/tests/test_log.py b/examples/raft/tests/test_log.py
index 6c15aaa..1108e34 100644
--- a/examples/raft/tests/test_log.py
+++ b/examples/raft/tests/test_log.py
@@ -76,7 +76,7 @@ def test_commits_and_applies_up_to_the_leaders_mark() -> None:
     append(node, entries=[[1, "a"], [1, "b"]], commit=1)
     assert node.commit_index == 1
     assert node.applied == [Entry(1, "a")]
-    assert events == [("apply", "n1", 1, 1, "a")]
+    assert events == [("apply", "n1", 1, 1, "a", 1)]
 
 
 def test_a_noop_advances_commit_but_not_the_state_machine() -> None:
@@ -85,7 +85,7 @@ def test_a_noop_advances_commit_but_not_the_state_machine() -> None:
     append(node, entries=[[1, ""], [1, "x"]], commit=2)
     assert node.commit_index == 2
     assert node.applied == [Entry(1, "x")]
-    assert events == [("apply", "n1", 2, 1, "x")]
+    assert events == [("apply", "n1", 2, 1, "x", 1)]
 
 
 def test_never_commits_past_what_it_verified() -> None: