diff --git a/test/jepsen/Dockerfile.node b/test/jepsen/Dockerfile.node
index 5f1107e..6439f42 100644
--- a/test/jepsen/Dockerfile.node
+++ b/test/jepsen/Dockerfile.node
@@ -11,6 +11,7 @@ ENV MIX_ENV=prod
 COPY mix.exs mix.lock ./
 COPY lib ./lib
 COPY test/jepsen/node.exs ./test/jepsen/node.exs
+COPY test/jepsen/repair_coverage.exs ./test/jepsen/repair_coverage.exs
 COPY test/support/test_tcp_transport.ex ./test/support/test_tcp_transport.ex
 
 RUN mix local.hex --force \
diff --git a/test/jepsen/README.md b/test/jepsen/README.md
index f0fb427..763c223 100644
--- a/test/jepsen/README.md
+++ b/test/jepsen/README.md
@@ -56,9 +56,65 @@ cannot masquerade as a current owner. Every history explicitly restarts one
 node after the deterministic conflict prelude, proving the checker does not
 mistake restart-sensitive instrumentation for missing protocol coverage.
 
+Registry-conflict exits are obligations, not trusted coverage counters. Before
+calling Group, each owner journals its registration attempt independently of
+Group's tables; it journals successful or rejected replies, unregisters, and
+cluster-intent removal as well. Drivers retain the victim token, key, and
+winner metadata from every conflict death. The checker reconstructs the
+victim's registrations, requires the reported winner to rank strictly higher
+by `{revision, token}`, and requires a matching historical winning claim in
+the same cluster and key. Only validated deaths count toward coverage.
+
+Registration calls interrupted by death or an indeterminate reply remain
+possible claims: Group may have installed them before the owner could record
+success. A definitive `:taken` reply excludes an attempt. Winning evidence is
+retained after unregister, death, and BEAM restart because delayed replicas
+can legitimately act on an older claim. Local journal order excludes winners
+first attempted after a death; the oracle does not invent a global clock or
+infer remote deletion delivery from wall time. This establishes independently
+witnessed possible winners, not the exact instant a replica learned a claim.
+
+The append-only conflict journal retains small operation records for the
+bounded campaign, not production ETS rows. Its path defaults to
+`/tmp/group-jepsen-conflict-evidence` inside each container and can be overridden
+with the driver's `:conflict_evidence_path` option. Corrupt or unreadable
+journals fail closed. At permanent retirement, the stopped-container collector
+archives and decodes the journal alongside unexpected deaths. The checker
+replays retired and surviving nodes' evidence together, without treating retired
+owners as live. Conflict archives are bounded to 64 MiB with ten-second command
+deadlines; missing or malformed archives fail the history.
+
+Each new history resets the running recorder through the harness socket after
+DB restart and before workload mutations. This clears disk and in-memory
+evidence together and initializes an empty journal, so repeated histories
+cannot inherit earlier conflict coverage.
+
+Checker qualification also loads the real Elixir
+Owner/Driver harness, injects valid and forged death reasons, exercises a
+register interrupted before its reply, and checks the emitted EDN with the
+Clojure lifecycle oracle.
+
+Repair coverage is receiver evidence, not transport admission. Before starting
+Group, the Jepsen VM compiles test-only wrappers around the existing delta-run
+and terminal snapshot-install functions. Each wrapper samples the same stream's
+cursor before and after its original body in one shard turn. Only positive
+committed advancement emits `applied-delta-run-records-peak`; a completed snapshot
+with a multi-chunk manifest emits `multi-chunk-snapshot-committed`. Duplicate
+prefixes do not contribute to the delta peak. Provisional chunks, missing
+terminal frames, rejected authority, and logical drops contribute neither.
+The wrapper installation fails if either expected boundary disappears.
+
+The receiver evidence owner persists increasing maxima outside the shard and
+reloads them after VM restart. A crash before persistence can lose evidence
+(failing coverage conservatively), but cannot create it. This instrumentation
+is shared by distribution, TCP, and chaos and changes no production BEAM or API.
+Sender `attempted-*` peaks remain diagnostics only. The pure checker reads the
+same EDN fixture that executable receiver regressions compare to live output.
+
 ## Requirements
 
 - Docker with Compose v2
+- Elixir/Mix with the repository dependencies installed (checker qualification)
 - Java 21 or newer
 - `curl`
 
diff --git a/test/jepsen/conflict_probe.exs b/test/jepsen/conflict_probe.exs
new file mode 100644
index 0000000..f276ff6
--- /dev/null
+++ b/test/jepsen/conflict_probe.exs
@@ -0,0 +1,146 @@
+# Invoked by the Clojure checker qualification. Load the actual harness without
+# starting its TCP server; no copied Owner/Driver logic lives in this probe.
+System.put_env("GROUP_JEPSEN_LIBRARY_ONLY", "1")
+Code.require_file("node.exs", __DIR__)
+Application.ensure_all_started(:group)
+Logger.configure(level: :emergency)
+
+defmodule Group.Jepsen.ConflictProbe do
+  alias Group.Jepsen.{ConflictEvidence, Driver, EDN}
+
+  def run do
+    directory = Path.join(__DIR__, ".cache/conflict-probe-#{System.unique_integer([:positive])}")
+    File.mkdir_p!(directory)
+
+    try do
+      {winner, rejected, winner_events, archive} = winner(directory)
+
+      cases = [
+        {"historical winner subsequently unregistered and died", true, 1, "jepsen/registry/0",
+         winner, false},
+        {"victim killed during register", true, 1, "jepsen/registry/0", winner, true},
+        {"forged key and rank", false, 99, "nonexistent", %{token: "invented", revision: -100},
+         false},
+        {"rightful winner killed", false, 99, "jepsen/registry/0", winner, false},
+        {"nonexistent winning claim", false, 1, "jepsen/registry/0",
+         %{token: "invented", revision: 100}, false},
+        {"definitively rejected winning claim", false, 1, "jepsen/registry/0", rejected, false}
+      ]
+
+      scenarios =
+        Enum.with_index(cases, fn {label, valid, revision, key, meta, pending}, index ->
+          {events, reset_events} = victim(directory, index, revision, key, meta, pending)
+
+          %{
+            label: label,
+            valid: valid,
+            archive: archive,
+            reset_evidence: reset_events,
+            snapshots: %{
+              "n1" => %{conflict_evidence: events},
+              "n2" => %{conflict_evidence: winner_events}
+            }
+          }
+        end)
+
+      IO.puts("CONFLICT-PROBE " <> EDN.encode(scenarios))
+    after
+      File.rm_rf!(directory)
+    end
+  end
+
+  defp start(directory, label, node_id) do
+    {:ok, group} = Group.start_link(name: :jepsen_group, shards: 1, log: false)
+    path = Path.join(directory, label)
+    {:ok, evidence} = ConflictEvidence.start_link(conflict_evidence_path: path)
+    {:ok, driver} = Driver.start_link(index: 0, node_id: node_id, boot_id: label)
+    {group, evidence, driver, path}
+  end
+
+  defp mutate(driver, operation, revision) do
+    GenServer.call(driver, {:mutate, operation, "owner", nil, 0, revision})
+  end
+
+  defp winner(directory) do
+    {group, evidence, driver, path} = start(directory, "winner", "n2")
+    %{status: :ok, owner: %{token: token}} = mutate(driver, :register, 10)
+
+    %{status: :fail, owner: %{token: rejected}} =
+      GenServer.call(driver, {:mutate, :register, "rejected", nil, 0, 100})
+
+    %{status: :ok} = mutate(driver, :unregister, 0)
+    %{status: :ok} = GenServer.call(driver, {:kill, "owner"})
+    %{status: :ok} = GenServer.call(driver, {:kill, "rejected"})
+    events = ConflictEvidence.snapshot()
+    archive = File.read!(path)
+    GenServer.stop(driver)
+    GenServer.stop(evidence)
+    Supervisor.stop(group)
+    {%{token: token, revision: 10}, %{token: rejected, revision: 100}, events, archive}
+  end
+
+  defp victim(directory, index, revision, key, winner, pending) do
+    {group, evidence, driver, path} = start(directory, "victim-#{index}", "n1")
+    %{status: :ok} = mutate(driver, :join, 0)
+    {pid, _token, _monitor, _cached} = :sys.get_state(driver).owners["owner"]
+    shard = Group.Replica.shard_for(:jepsen_group, nil, "jepsen/registry/0")
+
+    task =
+      if pending do
+        :ok = :sys.suspend(shard)
+        task = Task.async(fn -> mutate(driver, :register, revision) end)
+        wait(fn -> Enum.any?(ConflictEvidence.snapshot(), &(&1.kind == :register)) end)
+        task
+      else
+        %{status: :ok} = mutate(driver, :register, revision)
+        nil
+      end
+
+    Process.exit(pid, {:group_registry_conflict, key, winner})
+    if task, do: Task.await(task)
+    wait(fn -> Enum.any?(ConflictEvidence.snapshot(), &(&1.kind == :death)) end)
+    if pending, do: :sys.resume(shard)
+    events = ConflictEvidence.snapshot()
+    [] = GenServer.call(driver, :unexpected_deaths)
+    {:ok, []} = GenServer.call(driver, :owner_snapshots)
+    GenServer.stop(driver)
+    GenServer.stop(evidence)
+
+    # The exact registration/death obligations must survive a recorder restart.
+    {:ok, evidence} = ConflictEvidence.start_link(conflict_evidence_path: path)
+    ^events = ConflictEvidence.snapshot()
+    :ok = ConflictEvidence.reset()
+    [] = ConflictEvidence.snapshot()
+    "" = File.read!(path)
+    GenServer.stop(evidence)
+    {:ok, evidence} = ConflictEvidence.start_link(conflict_evidence_path: path)
+    [] = ConflictEvidence.snapshot()
+
+    1 =
+      ConflictEvidence.record(%{
+        kind: :register,
+        token: "next-history",
+        key: 0,
+        cluster: nil,
+        revision: 1
+      })
+
+    :ok = ConflictEvidence.reset()
+    reset_events = ConflictEvidence.snapshot()
+    GenServer.stop(evidence)
+    Supervisor.stop(group)
+    {events, reset_events}
+  end
+
+  defp wait(fun, remaining \\ 200)
+  defp wait(_fun, 0), do: raise("probe timed out")
+
+  defp wait(fun, remaining) do
+    unless fun.() do
+      Process.sleep(5)
+      wait(fun, remaining - 1)
+    end
+  end
+end
+
+Group.Jepsen.ConflictProbe.run()
diff --git a/test/jepsen/decode_conflict_evidence.exs b/test/jepsen/decode_conflict_evidence.exs
new file mode 100644
index 0000000..0fcc110
--- /dev/null
+++ b/test/jepsen/decode_conflict_evidence.exs
@@ -0,0 +1,6 @@
+System.put_env("GROUP_JEPSEN_LIBRARY_ONLY", "1")
+Code.require_file("node.exs", __DIR__)
+
+[path] = System.argv()
+events = path |> File.read!() |> Group.Jepsen.ConflictEvidence.decode()
+IO.puts("CONFLICT-EVIDENCE " <> Group.Jepsen.EDN.encode(events))
diff --git a/test/jepsen/fixtures/applied-repair-events.edn b/test/jepsen/fixtures/applied-repair-events.edn
new file mode 100644
index 0000000..e49d46e
--- /dev/null
+++ b/test/jepsen/fixtures/applied-repair-events.edn
@@ -0,0 +1 @@
+{:applied-delta-run-records-peak 2 :multi-chunk-snapshot-committed 1}
diff --git a/test/jepsen/metadata_capture.exs b/test/jepsen/metadata_capture.exs
index 9f1d5db..2ad58ea 100644
--- a/test/jepsen/metadata_capture.exs
+++ b/test/jepsen/metadata_capture.exs
@@ -1,9 +1,17 @@
 System.put_env("GROUP_JEPSEN_LIBRARY", "1")
+Code.require_file("repair_coverage.exs", __DIR__)
 Code.require_file("node.exs", __DIR__)
 
 alias Group.Jepsen.{Driver, EDN, Snapshot}
 alias Group.Replica.Data
 
+repair_path =
+  Path.join(
+    System.tmp_dir!(),
+    "group-jepsen-metadata-repair-#{System.unique_integer([:positive])}"
+  )
+
+{:ok, _} = Group.Jepsen.RepairCoverage.start_link(path: repair_path)
 {:ok, _} = Group.Jepsen.Transport.Stats.start_link([])
 
 {:ok, _} =
@@ -74,3 +82,4 @@ Enum.each(originals, fn {table, rows} -> :ets.insert(table, rows) end)
 false = reincarnated.token == owner.token
 
 File.write!(hd(System.argv()), EDN.encode(%{healthy: healthy, corruptions: corruptions}))
+File.rm(repair_path)
diff --git a/test/jepsen/node.exs b/test/jepsen/node.exs
index bb1b6aa..f4cbae5 100644
--- a/test/jepsen/node.exs
+++ b/test/jepsen/node.exs
@@ -1,4 +1,5 @@
 Code.require_file("../support/test_tcp_transport.ex", __DIR__)
+Code.require_file("repair_coverage.exs", __DIR__)
 
 defmodule Group.Jepsen.Transport.Stats do
   @moduledoc false
@@ -38,6 +39,7 @@ defmodule Group.Jepsen.Transport.Stats do
     |> :ets.tab2list()
     |> Map.new()
     |> Map.merge(persisted, fn _event, current, durable -> max(current, durable) end)
+    |> Map.merge(Group.Jepsen.RepairCoverage.snapshot())
   end
 
   def block(target_node), do: :ets.insert(@gate, {target_node})
@@ -103,9 +105,7 @@ defmodule Group.Jepsen.Transport.Common do
   def record({:snapshot_commit, _version, _stream, _seq, chunk_count, _, _}) do
     Stats.increment(:snapshot_commit)
 
-    if chunk_count > 1 do
-      Stats.increment(:multi_chunk_snapshot)
-    end
+    Stats.observe_max(:attempted_snapshot_chunks_peak, chunk_count)
   end
 
   def record({:delta_batch, _version, runs}) do
@@ -119,7 +119,7 @@ defmodule Group.Jepsen.Transport.Common do
       end)
       |> Enum.max(fn -> 0 end)
 
-    Stats.observe_max(:delta_run_records_peak, peak)
+    Stats.observe_max(:attempted_delta_run_records_peak, peak)
   end
 
   def record(_message), do: Stats.increment(:other_message)
@@ -424,6 +424,73 @@ defmodule Group.Jepsen.ConflictResolver do
   defp rank(_meta), do: {-1, ""}
 end
 
+defmodule Group.Jepsen.ConflictEvidence do
+  @moduledoc false
+  use GenServer
+
+  # Independent of Group's ETS, and retained across container/BEAM restarts.
+  # Persist the invocation before calling Group: a conflict can kill an owner
+  # before its register call returns, even though its claim was installed.
+  def start_link(opts), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__)
+  def record(event), do: GenServer.call(__MODULE__, {:record, event})
+  def snapshot, do: GenServer.call(__MODULE__, :snapshot)
+  def reset, do: GenServer.call(__MODULE__, :reset)
+
+  def decode(contents) do
+    if contents != "" and not String.ends_with?(contents, "\n"),
+      do: raise("truncated conflict evidence")
+
+    contents
+    |> String.split("\n")
+    |> Enum.drop(-1)
+    |> Enum.with_index(1)
+    |> Enum.map(fn {line, sequence} ->
+      event = line |> Base.decode64!() |> :erlang.binary_to_term()
+      %{sequence: ^sequence} = event
+      event
+    end)
+  end
+
+  @impl true
+  def init(opts) do
+    path = Keyword.get(opts, :conflict_evidence_path, "/tmp/group-jepsen-conflict-evidence")
+
+    events =
+      case File.read(path) do
+        {:ok, contents} ->
+          decode(contents)
+
+        {:error, :enoent} ->
+          []
+
+        {:error, reason} ->
+          raise "cannot read conflict evidence: #{inspect(reason)}"
+      end
+
+    {:ok, %{path: path, events: Enum.reverse(events), sequence: length(events)}}
+  end
+
+  @impl true
+  def handle_call({:record, event}, _from, state) do
+    event = Map.put(event, :sequence, state.sequence + 1)
+    encoded = event |> :erlang.term_to_binary() |> Base.encode64()
+    :ok = File.write(state.path, encoded <> "\n", [:append, :sync])
+    {:reply, event.sequence, %{state | events: [event | state.events], sequence: event.sequence}}
+  end
+
+  def handle_call(:snapshot, _from, state) do
+    {:reply, Enum.reverse(state.events), state}
+  end
+
+  # Called only during DB setup, after restart and before workload mutations.
+  # Removing the file externally would leave the restarted recorder's loaded
+  # evidence alive in memory and leak coverage into the next history.
+  def handle_call(:reset, _from, state) do
+    :ok = File.write(state.path, "", [:sync])
+    {:reply, :ok, %{state | events: [], sequence: 0}}
+  end
+end
+
 defmodule Group.Jepsen.Owner do
   @moduledoc false
   use GenServer
@@ -437,15 +504,26 @@ defmodule Group.Jepsen.Owner do
   def handle_call({:mutate, :register, cluster, key, revision}, _from, state) do
     meta = %{token: state.token, revision: revision}
 
+    attempt =
+      Group.Jepsen.ConflictEvidence.record(%{
+        kind: :register,
+        token: state.token,
+        cluster: cluster,
+        key: key,
+        revision: revision
+      })
+
     case safe_group_call(fn ->
            Group.register(:jepsen_group, registry_key(key), meta, cluster_opts(cluster))
          end) do
       :ok ->
+        registration_result(attempt, :ok)
         entry = %{cluster: cluster, key: key, revision: revision}
         state = put_in(state.registrations[{cluster, key}], entry)
         {:reply, {:ok, snapshot(state)}, state}
 
       {:error, reason} ->
+        registration_result(attempt, if(reason == :taken, do: :fail, else: :unknown))
         {:reply, {:error, reason, snapshot(state)}, state}
     end
   end
@@ -458,6 +536,13 @@ defmodule Group.Jepsen.Owner do
              Group.unregister(:jepsen_group, registry_key(key), cluster_opts(cluster))
            end) do
         :ok ->
+          Group.Jepsen.ConflictEvidence.record(%{
+            kind: :unregister,
+            token: state.token,
+            cluster: cluster,
+            key: key
+          })
+
           state = %{state | registrations: Map.delete(state.registrations, owner_key)}
           {:reply, {:ok, snapshot(state)}, state}
 
@@ -505,6 +590,12 @@ defmodule Group.Jepsen.Owner do
   end
 
   def handle_call({:drop_cluster, cluster}, _from, state) do
+    Group.Jepsen.ConflictEvidence.record(%{
+      kind: :drop_cluster,
+      token: state.token,
+      cluster: cluster
+    })
+
     registrations = drop_cluster(state.registrations, cluster)
     memberships = drop_cluster(state.memberships, cluster)
     state = %{state | registrations: registrations, memberships: memberships}
@@ -513,6 +604,10 @@ defmodule Group.Jepsen.Owner do
 
   def handle_call(:snapshot, _from, state), do: {:reply, snapshot(state), state}
 
+  defp registration_result(attempt, status) do
+    Group.Jepsen.ConflictEvidence.record(%{kind: :result, attempt: attempt, status: status})
+  end
+
   defp drop_cluster(entries, cluster) do
     entries
     |> Enum.reject(fn {{entry_cluster, _key}, _entry} -> entry_cluster == cluster end)
@@ -547,8 +642,6 @@ defmodule Group.Jepsen.Driver do
   @moduledoc false
   use GenServer
 
-  alias Group.Jepsen.Transport.Stats
-
   @driver_count 8
   @unexpected_death_log "/tmp/group-jepsen-unexpected-deaths"
 
@@ -731,7 +824,16 @@ defmodule Group.Jepsen.Driver do
 
         unexpected_deaths =
           if match?({:group_registry_conflict, _key, _winner_meta}, reason) do
-            Stats.increment_persistent(:registry_conflict_death)
+            {:group_registry_conflict, key, winner_meta} = reason
+
+            Group.Jepsen.ConflictEvidence.record(%{
+              kind: :death,
+              token: token,
+              key: if(is_binary(key), do: key, else: %{invalid: inspect(key)}),
+              winner:
+                if(is_map(winner_meta), do: winner_meta, else: %{invalid: inspect(winner_meta)})
+            })
+
             state.unexpected_deaths
           else
             death = %{token: token, reason: inspect(reason)}
@@ -821,7 +923,11 @@ defmodule Group.Jepsen.Driver.Supervisor do
 
   @impl true
   def init(opts),
-    do: Supervisor.init(Group.Jepsen.Driver.child_specs(opts), strategy: :one_for_one)
+    do:
+      Supervisor.init(
+        [{Group.Jepsen.ConflictEvidence, opts} | Group.Jepsen.Driver.child_specs(opts)],
+        strategy: :one_for_one
+      )
 end
 
 defmodule Group.Jepsen.Cluster do
@@ -1308,6 +1414,7 @@ defmodule Group.Jepsen.Snapshot do
         peers: Group.nodes(:jepsen_group) |> Enum.map(&Atom.to_string/1) |> Enum.sort(),
         owners: owners,
         unexpected_deaths: Group.Jepsen.Driver.unexpected_deaths(),
+        conflict_evidence: Group.Jepsen.ConflictEvidence.snapshot(),
         transport_events: Group.Jepsen.Transport.Stats.snapshot(),
         transport_profile: Group.Jepsen.Transport.Control.profile(),
         internal: Group.Jepsen.Invariant.snapshot(retired_nodes),
@@ -1390,6 +1497,10 @@ defmodule Group.Jepsen.Wire do
       ["ping"] ->
         %{status: :ok}
 
+      ["reset-conflict-evidence"] ->
+        :ok = Group.Jepsen.ConflictEvidence.reset()
+        %{status: :ok}
+
       ["ready", expected] ->
         expected = String.to_integer(expected)
 
@@ -1551,6 +1662,11 @@ defmodule Group.Jepsen.Main do
 
     {:ok, _apps} = Application.ensure_all_started(:group)
 
+    :ok = Group.Jepsen.RepairCoverage.install!()
+
+    {:ok, _coverage} =
+      Group.Jepsen.RepairCoverage.start_link(path: "/tmp/group-jepsen-repair-coverage")
+
     {:ok, group} =
       Group.start_link(
         name: :jepsen_group,
@@ -1596,6 +1712,7 @@ defmodule Group.Jepsen.Main do
   end
 end
 
-unless System.get_env("GROUP_JEPSEN_LIBRARY") == "1" do
+unless System.get_env("GROUP_JEPSEN_LIBRARY") == "1" or
+         System.get_env("GROUP_JEPSEN_LIBRARY_ONLY") == "1" do
   Group.Jepsen.Main.run(System.argv())
 end
diff --git a/test/jepsen/repair_coverage.exs b/test/jepsen/repair_coverage.exs
new file mode 100644
index 0000000..6f0df94
--- /dev/null
+++ b/test/jepsen/repair_coverage.exs
@@ -0,0 +1,163 @@
+defmodule Group.Jepsen.RepairCoverage do
+  @moduledoc """
+  Test-only instrumentation of receiver commit boundaries. No sender admission,
+  socket completion, or provisional snapshot chunk is evidence of application.
+
+  Compile before starting Group. Wrapping the two private functions keeps the
+  cursor observations in the receiving shard's serialized turn; an unrelated
+  run cannot advance the cursor between them. Production BEAMs are unchanged.
+  """
+  use GenServer
+
+  def install! do
+    path = Path.expand("../../lib/group/replica.ex", __DIR__)
+    ast = path |> File.read!() |> Code.string_to_quoted!(file: path)
+
+    {ast, wrapped} =
+      Macro.postwalk(ast, [], fn
+        {:defp, meta, [{name, _, args} = signature, [do: body]]}, found
+        when name in [:apply_replica_delta_run, :commit_snapshot_transfer] ->
+          expected =
+            case name do
+              :apply_replica_delta_run ->
+                [:state, :source_node, :stream_id, :records, :advertised_head]
+
+              :commit_snapshot_transfer ->
+                [:state, :key, :source_node, :stream_id, :transfer]
+            end
+
+          unless Enum.map(args, &elem(&1, 0)) == expected do
+            raise "replica repair signature changed: #{Macro.to_string(signature)}"
+          end
+
+          [state, _, stream | _] =
+            if name == :commit_snapshot_transfer do
+              [Enum.at(args, 0), Enum.at(args, 2), Enum.at(args, 3)]
+            else
+              args
+            end
+
+          kind =
+            if name == :commit_snapshot_transfer,
+              do: quote(do: {:snapshot, elem(unquote(List.last(args)).manifest, 0)}),
+              else: :delta
+
+          wrapped =
+            quote do
+              before_cursor =
+                Group.Replica.Data.replica_cursor(
+                  unquote(state).name,
+                  unquote(state).shard_index,
+                  unquote(stream)
+                )
+
+              result = unquote(body)
+
+              after_cursor =
+                Group.Replica.Data.replica_cursor(
+                  unquote(state).name,
+                  unquote(state).shard_index,
+                  unquote(stream)
+                )
+
+              Group.Jepsen.RepairCoverage.observe(
+                unquote(kind),
+                before_cursor,
+                after_cursor
+              )
+
+              result
+            end
+
+          {{:defp, meta, [signature, [do: wrapped]]}, [name | found]}
+
+        other, found ->
+          {other, found}
+      end)
+
+    unless Enum.sort(wrapped) == [:apply_replica_delta_run, :commit_snapshot_transfer] do
+      raise "replica repair boundaries changed: #{inspect(wrapped)}"
+    end
+
+    Code.compile_quoted(ast, path)
+    :ok
+  end
+
+  def start_link(opts), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__)
+
+  # Only a local message on the shard. Disk I/O happens in this test-only owner.
+  def observe(kind, before_cursor, after_cursor) when after_cursor > before_cursor do
+    GenServer.cast(__MODULE__, {:committed, kind, after_cursor - before_cursor})
+  end
+
+  def observe(_, _, _), do: :ok
+
+  def snapshot, do: GenServer.call(__MODULE__, :snapshot)
+
+  @impl true
+  def init(opts) do
+    path = Keyword.fetch!(opts, :path)
+    {:ok, %{path: path, events: read_events(path)}}
+  end
+
+  @impl true
+  def handle_cast({:committed, kind, records}, state) do
+    evidence =
+      case kind do
+        :delta -> %{applied_delta_run_records_peak: records}
+        {:snapshot, chunks} when chunks > 1 -> %{multi_chunk_snapshot_committed: 1}
+        {:snapshot, _} -> %{}
+      end
+
+    previous = if File.exists?(state.path), do: state.events, else: %{}
+    events = Map.merge(previous, evidence, fn _, old, new -> max(old, new) end)
+
+    if events != previous do
+      # Append only this observation, never cached evidence from a previous
+      # history. reset-oracle! deletes the file while the VM may already be up.
+      for {event, value} <- evidence do
+        File.write!(state.path, "#{event}\t#{value}\n", [:append])
+      end
+    end
+
+    {:noreply, %{state | events: events}}
+  end
+
+  @impl true
+  def handle_call(:snapshot, _from, state) do
+    events = read_events(state.path)
+    {:reply, events, %{state | events: events}}
+  end
+
+  defp read_events(path) do
+    case File.read(path) do
+      {:ok, contents} ->
+        {committed, [suffix]} = contents |> String.split("\n") |> Enum.split(-1)
+
+        events =
+          Enum.reduce(committed, %{}, fn line, events ->
+            [event, value] = String.split(line, "\t")
+            event = String.to_existing_atom(event)
+            value = String.to_integer(value)
+            Map.update(events, event, value, &max(&1, value))
+          end)
+
+        # Validate committed records before touching the file. An interrupted
+        # append is not evidence, and must not become the next append's prefix.
+        # Truncate only the suffix in place, preserving every committed byte.
+        if suffix != "" do
+          offset = byte_size(contents) - byte_size(suffix)
+
+          File.open!(path, [:read, :write, :binary], fn file ->
+            {:ok, ^offset} = :file.position(file, offset)
+            :ok = :file.truncate(file)
+          end)
+        end
+
+        events
+
+      {:error, :enoent} ->
+        %{}
+    end
+  end
+end
diff --git a/test/jepsen/src/group/jepsen/db.clj b/test/jepsen/src/group/jepsen/db.clj
index 702b41c..8898077 100644
--- a/test/jepsen/src/group/jepsen/db.clj
+++ b/test/jepsen/src/group/jepsen/db.clj
@@ -9,6 +9,10 @@
     (docker/heal! (:nodes test))
     (docker/restart! node)
     (docker/reset-oracle! node)
+    (group-client/wait-listening! node)
+    (let [response (group-client/request! node ["reset-conflict-evidence"])]
+      (when-not (= :ok (:status response))
+        (throw (ex-info "conflict oracle reset failed" {:node node :response response}))))
     (group-client/wait-ready! node (count (:nodes test))))
 
   (teardown! [_this test _node]
diff --git a/test/jepsen/src/group/jepsen/docker.clj b/test/jepsen/src/group/jepsen/docker.clj
index d0fc147..6c83bc5 100644
--- a/test/jepsen/src/group/jepsen/docker.clj
+++ b/test/jepsen/src/group/jepsen/docker.clj
@@ -1,5 +1,6 @@
 (ns group.jepsen.docker
-  (:require [clojure.string :as str])
+  (:require [clojure.edn :as edn]
+            [clojure.string :as str])
   (:import (java.io File)
            (java.util.concurrent TimeUnit)))
 
@@ -87,6 +88,7 @@
 (defn reset-oracle! [node]
   (exec-sh! node
             (str "rm -f /tmp/group-jepsen-persistent-events "
+                 "/tmp/group-jepsen-repair-coverage "
                  "/tmp/group-jepsen-cursor-marker-corruption && "
                  ": > /tmp/group-jepsen-unexpected-deaths")))
 
@@ -98,24 +100,45 @@
             {:token token :reason reason}))
         (remove str/blank? (str/split-lines contents))))
 
-(defn retired-evidence!
-  "Reads the stopped container's durable oracle, without depending on its VM or socket.
-  Missing, truncated, unreadable, or oversized evidence is a qualification failure."
-  [node]
+(defn decode-conflict-evidence! [file]
+  (let [output (shell! "sh" "-c"
+                       (str "cd ../.. && exec env ERL_FLAGS='+S 2:2' mix run --no-start "
+                            "test/jepsen/decode_conflict_evidence.exs \"$1\"")
+                       "_" (.getAbsolutePath ^File file))
+        prefix "CONFLICT-EVIDENCE "
+        line (first (filter #(str/starts-with? % prefix) (str/split-lines output)))]
+    (when-not line
+      (throw (ex-info "missing decoded conflict evidence" {})))
+    (edn/read-string (subs line (count prefix)))))
+
+(defn collect-evidence! [node path bound decode]
   (let [file (File/createTempFile "group-jepsen-retired-" ".log")]
     (try
       (binding [*command-timeout-ms* 10000]
         (docker! "cp"
-                 (str (container node) ":/tmp/group-jepsen-unexpected-deaths")
+                 (str (container node) ":" path)
                  (.getAbsolutePath file)))
-      (when (> (.length file) (* 8 1024 1024))
+      (when (> (.length file) bound)
         (throw (ex-info "lifecycle evidence exceeds collection bound" {:node node})))
       (let [contents (slurp file)]
         (when (and (seq contents) (not (str/ends-with? contents "\n")))
           (throw (ex-info "truncated lifecycle evidence" {:node node})))
-        {:node (name node) :unexpected-deaths (parse-unexpected-deaths contents)})
+        (binding [*command-timeout-ms* 10000]
+          (decode file)))
       (finally (.delete file)))))
 
+(defn retired-evidence!
+  "Reads the stopped container's durable oracle, without depending on its VM or socket.
+  Missing, truncated, unreadable, or oversized evidence is a qualification failure."
+  [node]
+  {:node (name node)
+   :unexpected-deaths
+   (collect-evidence! node "/tmp/group-jepsen-unexpected-deaths" (* 8 1024 1024)
+                      #(parse-unexpected-deaths (slurp %)))
+   :conflict-evidence
+   (collect-evidence! node "/tmp/group-jepsen-conflict-evidence" (* 64 1024 1024)
+                      decode-conflict-evidence!)})
+
 (defn ensure-firewall-chain! [node chain]
   (exec-sh!
     node
diff --git a/test/jepsen/src/group/jepsen/model.clj b/test/jepsen/src/group/jepsen/model.clj
index 7c5da84..0426743 100644
--- a/test/jepsen/src/group/jepsen/model.clj
+++ b/test/jepsen/src/group/jepsen/model.clj
@@ -4,7 +4,7 @@
             [jepsen.history :as history]))
 
 (def default-required-transport-events
-  #{:delta-batch :snapshot-chunk :multi-chunk-snapshot
+  #{:applied-delta-run-records-peak :multi-chunk-snapshot-committed
     :registry-conflict-death})
 
 (defn successful-snapshots [history]
@@ -90,6 +90,7 @@
   {:owners (set (:owners snapshot))
    :peers (set (:peers snapshot))
    :unexpected-deaths (set (:unexpected-deaths snapshot))
+   :conflict-evidence (:conflict-evidence snapshot)
    :view (normalize-view test snapshot)
    :internal (stable-internal snapshot)})
 
@@ -98,6 +99,79 @@
        (remove history/invoke?)
        (keep #(get-in % [:value :response :latency-us]))))
 
+(defn replay-conflict-evidence [node events]
+  (reduce
+    (fn [state {:keys [kind sequence token cluster key attempt status] :as event}]
+      (let [slot [cluster key]]
+        (case kind
+          :register (-> state
+                        (assoc-in [:claims sequence] (assoc event :node node))
+                        (assoc-in [:pending token sequence] slot))
+          :result (let [claim (get-in state [:claims attempt])
+                        token (:token claim)
+                        slot [(:cluster claim) (:key claim)]]
+                    (cond-> (assoc-in state [:claims attempt :status] status)
+                      (not= :unknown status) (update-in [:pending token] dissoc attempt)
+                      (= :ok status) (assoc-in [:active token slot] attempt)))
+          :unregister
+          (-> state
+              (update-in [:active token] dissoc slot)
+              (update-in [:pending token]
+                         #(into {} (remove (fn [[_ s]] (= slot s))) %)))
+          :drop-cluster
+          (-> state
+              (update-in [:active token]
+                         #(into {} (remove (fn [[[c _] _]] (= cluster c))) %))
+              (update-in [:pending token]
+                         #(into {} (remove (fn [[_ [c _]]] (= cluster c))) %)))
+          :death
+          (let [attempts (concat (vals (get-in state [:active token]))
+                                 (keys (get-in state [:pending token])))]
+            (-> state
+                (update :deaths conj
+                        (assoc event :node node :victim-attempts (set attempts)))
+                (update :active dissoc token)
+                (update :pending dissoc token)))
+          state)))
+    {:claims {} :active {} :pending {} :deaths []}
+    events))
+
+(defn conflict-analysis [snapshots]
+  (let [journals (into {} (map (fn [[node snapshot]]
+                                [node (replay-conflict-evidence
+                                        node (:conflict-evidence snapshot))]))
+                       snapshots)
+        claims (mapcat (comp vals :claims val) journals)
+        possible? #(not= :fail (:status %))
+        winners (group-by (juxt :token :revision :key) (filter possible? claims))
+        deaths (mapcat (comp :deaths val) journals)
+        justified?
+        (fn [{:keys [node sequence token key winner victim-attempts]}]
+          (boolean
+            (some
+              (fn [id]
+                (let [victim (get-in journals [node :claims id])
+                      rank (juxt :revision :token)]
+                  (and (possible? victim)
+                       (= token (:token victim))
+                       (= key (str "jepsen/registry/" (:key victim)))
+                       (integer? (:revision winner))
+                       (string? (:token winner))
+                       (not= token (:token winner))
+                       (pos? (compare (rank winner) (rank victim)))
+                       (some #(and (= (:cluster victim) (:cluster %))
+                                   ;; Local order is known. Across nodes there
+                                   ;; is no synchronized oracle clock; delayed
+                                   ;; remote deletions may still lose a conflict.
+                                   (or (not= node (:node %))
+                                       (< (:sequence %) sequence)))
+                             (get winners [(:token winner) (:revision winner)
+                                           (:key victim)])))))
+              victim-attempts)))
+        invalid (vec (remove justified? deaths))]
+    {:invalid invalid
+     :validated-count (- (count deaths) (count invalid))}))
+
 (defn analyze [test history]
   (let [observations (snapshots-by-node history)
         snapshots (latest-snapshots history)
@@ -139,10 +213,27 @@
                         (when (not= expected-peers actual-peers)
                           [node {:expected expected-peers, :actual actual-peers}]))))
               relevant-snapshots)
+        completions (remove history/invoke? history)
+        retirement-evidence (keep #(get-in % [:value :lifecycle-evidence]) completions)
+        ;; Retired nodes no longer contribute live owners or public views, but
+        ;; their durable claims and deaths remain obligations for this history.
+        conflict-snapshots
+        (into {}
+              (map (fn [[node evidence]]
+                     [node {:conflict-evidence (->> evidence
+                                                   (mapcat :conflict-evidence)
+                                                   distinct
+                                                   (sort-by :sequence)
+                                                   vec)}]))
+              (group-by :node
+                        (concat (map :value (successful-snapshots history))
+                                retirement-evidence)))
+        conflicts (conflict-analysis conflict-snapshots)
         transport-events
-        (reduce #(merge-with + %1 %2)
-                {}
-                (map #(or (:transport-events %) {}) (vals relevant-snapshots)))
+        (assoc (reduce #(merge-with + %1 %2)
+                       {}
+                       (map #(or (:transport-events %) {}) (vals relevant-snapshots)))
+               :registry-conflict-death (:validated-count conflicts))
         required-transport-events
         (get test :required-transport-events
              default-required-transport-events)
@@ -150,7 +241,7 @@
         (set (remove #(pos? (get transport-events % 0)) required-transport-events))
         delta-run-records-peak
         (reduce max 0
-                (map #(get-in % [:transport-events :delta-run-records-peak] 0)
+                (map #(get-in % [:transport-events :applied-delta-run-records-peak] 0)
                      (vals relevant-snapshots)))
         min-delta-run-records (get test :min-delta-run-records 0)
         delta-run-coverage? (>= delta-run-records-peak min-delta-run-records)
@@ -170,8 +261,6 @@
                                   (not= 0 (:snapshot-staging-count internal)))
                           [node internal]))))
               relevant-snapshots)
-        completions (remove history/invoke? history)
-        retirement-evidence (keep #(get-in % [:value :lifecycle-evidence]) completions)
         retired-nodes (set/difference (set (map name (:nodes test))) required-nodes)
         collected-nodes (set (map :node retirement-evidence))
         evidence-errors (vec (keep #(get-in % [:value :evidence-error]) completions))
@@ -210,6 +299,7 @@
                     (empty? (:conflicts expected))
                     (empty? mismatches)
                     (empty? unexpected-deaths)
+                    (empty? (:invalid conflicts))
                     (empty? evidence-errors)
                     (empty? missing-retirement-evidence)
                     (empty? orphaned)
@@ -233,6 +323,7 @@
      :live-registry-conflicts (:conflicts expected)
      :mismatched-views mismatches
      :unexpected-owner-deaths unexpected-deaths
+     :invalid-conflict-deaths (:invalid conflicts)
      :lifecycle-evidence-errors evidence-errors
      :missing-retirement-evidence missing-retirement-evidence
      :orphaned-owner-tokens orphaned
diff --git a/test/jepsen/test/group/jepsen/model_test.clj b/test/jepsen/test/group/jepsen/model_test.clj
index 98091fe..5ea9b78 100644
--- a/test/jepsen/test/group/jepsen/model_test.clj
+++ b/test/jepsen/test/group/jepsen/model_test.clj
@@ -1,7 +1,16 @@
 (ns group.jepsen.model-test
-  (:require [clojure.test :refer :all]
+  (:require [clojure.edn :as edn]
+            [clojure.test :refer :all]
+            [clojure.java.shell :as shell]
+            [clojure.string :as str]
+            [group.jepsen.docker :as docker]
             [group.jepsen.model :as model]))
 
+(def applied-repair-events
+  ;; The receiver regression compares this fixture byte-for-byte with live EDN
+  ;; emitted after real delta application and terminal snapshot installation.
+  (edn/read-string (slurp "fixtures/applied-repair-events.edn")))
+
 (def test-map
   {:nodes ["n1" "n2" "n3"]
    :terminal-nodes ["n1" "n2" "n3"]
@@ -79,6 +88,125 @@
         (is (false? (:valid? result)))
         (is (contains? (:mismatched-views result) "n2"))))))
 
+(def conflict-journal
+  [{:kind :register :sequence 1 :token "a" :cluster nil :key 0 :revision 1}
+   {:kind :result :sequence 2 :attempt 1 :status :ok}
+   {:kind :register :sequence 3 :token "b" :cluster nil :key 0 :revision 2}
+   {:kind :death :sequence 4 :token "a" :key "jepsen/registry/0"
+    :winner {:token "b" :revision 2}}
+   {:kind :result :sequence 5 :attempt 3 :status :ok}
+   {:kind :unregister :sequence 6 :token "b" :cluster nil :key 0}])
+
+(defn with-conflict-evidence [op]
+  (assoc-in op [:value :conflict-evidence] conflict-journal))
+
+(deftest validates-historical-conflicts-independently
+  (let [check #(model/conflict-analysis {"n1" {:conflict-evidence %}})
+        valid #(is (empty? (:invalid (check %))))
+        invalid #(is (= 1 (count (:invalid (check %)))))]
+    (valid conflict-journal)
+    (valid (vec (remove #(= 2 (:sequence %)) conflict-journal)))
+    (invalid (assoc-in conflict-journal [3 :key] "nonexistent"))
+    (invalid (assoc-in conflict-journal [3 :winner :revision] -100))
+    (invalid (assoc-in conflict-journal [3 :winner :token] "invented"))
+    (invalid (assoc-in conflict-journal [3 :winner :token] "a"))
+    (invalid (assoc-in conflict-journal [2 :cluster] "red"))
+    (invalid (assoc-in conflict-journal [4 :status] :fail))
+    (invalid (assoc-in conflict-journal [0 :revision] 99))
+    (invalid (vec (concat (subvec conflict-journal 0 3)
+                          [{:kind :unregister :sequence 4 :token "a" :cluster nil :key 0}]
+                          (subvec conflict-journal 3))))
+    (invalid (vec (concat (subvec conflict-journal 0 3)
+                          [{:kind :drop-cluster :sequence 4 :token "a" :cluster nil}]
+                          (subvec conflict-journal 3))))
+    (invalid (conj conflict-journal
+                   {:kind :register :sequence 7 :token "b" :cluster nil :key 0 :revision 2}
+                   {:kind :death :sequence 8 :token "b" :key "jepsen/registry/0"
+                    :winner {:token "a" :revision 1}}))
+    (invalid (vec (remove #(= :register (:kind %)) conflict-journal)))
+    (invalid (vec (concat (subvec conflict-journal 0 2)
+                          [(last conflict-journal)]
+                          (subvec conflict-journal 3 4)
+                          [(assoc (get conflict-journal 2) :sequence 5)])))
+    ;; Unique incarnation tokens break equal-revision ties without pid order.
+    (valid (assoc-in conflict-journal [0 :revision] 2))))
+
+(deftest conflict-evidence-survives-permanent-retirement
+  (let [test (assoc test-map :terminal-nodes ["n2" "n3"]
+                            :required-transport-events #{:registry-conflict-death})
+        winner [{:kind :register :sequence 1 :token "winner" :cluster nil :key 0 :revision 10}
+                {:kind :result :sequence 2 :attempt 1 :status :ok}]
+        victim [{:kind :register :sequence 1 :token "victim" :cluster nil :key 0 :revision 1}
+                {:kind :death :sequence 2 :token "victim" :key "jepsen/registry/0"
+                 :winner {:token "winner" :revision 10}}]
+        retired (fn [events] {:type :info :f :retire
+                              :value {:lifecycle-evidence {:node "n1" :unexpected-deaths []
+                                                            :conflict-evidence events}}})
+        terminal [(assoc-in (snapshot-op 1 "n2" ["n2" "n3"] [] (empty-registry) (empty-pg))
+                            [:value :conflict-evidence] victim)
+                  (snapshot-op 2 "n3" ["n2" "n3"] [] (empty-registry) (empty-pg))]
+        healthy (model/analyze test (conj terminal (retired winner)))
+        bad (model/analyze (assoc test :required-transport-events #{})
+                           [(retired (assoc-in conflict-journal [3 :winner :revision] -100))
+                            (snapshot-op 2 "n2" ["n2" "n3"] [] (empty-registry) (empty-pg))
+                            (snapshot-op 3 "n3" ["n2" "n3"] [] (empty-registry) (empty-pg))])]
+    (is (:valid? healthy))
+    (is (= 1 (get-in healthy [:transport-events :registry-conflict-death])))
+    (is (false? (:valid? bad)))
+    (is (= 1 (count (:invalid-conflict-deaths bad))))))
+
+(deftest conflict-statistics-alone-do-not-discharge-deaths
+  (let [history [(assoc-in (snapshot-op 1 "n1" [] (empty-registry) (empty-pg))
+                           [:value :transport-events] {:registry-conflict-death 99})
+                 (snapshot-op 2 "n2" [] (empty-registry) (empty-pg))
+                 (snapshot-op 3 "n3" [] (empty-registry) (empty-pg))]
+        result (model/analyze (assoc test-map :required-transport-events
+                                    #{:registry-conflict-death}) history)]
+    (is (false? (:valid? result)))
+    (is (= #{:registry-conflict-death} (:missing-transport-events result)))))
+
+(deftest checks-real-owner-and-driver-evidence
+  (let [{:keys [exit out err]}
+        (shell/sh "mix" "run" "--no-start" "test/jepsen/conflict_probe.exs"
+                  :dir "../.."
+                  :env (assoc (into {} (System/getenv)) "ERL_FLAGS" "+S 2:2"))
+        _ (is (zero? exit) (str out err))
+        line (first (filter #(str/starts-with? % "CONFLICT-PROBE ")
+                            (str/split-lines out)))
+        scenarios (when line (edn/read-string (subs line 15)))]
+    (is (some? scenarios) (str out err))
+    (doseq [{:keys [label valid snapshots reset-evidence]} scenarios]
+      (let [history (mapv (fn [index node]
+                            (assoc-in (snapshot-op index node [] (empty-registry) (empty-pg))
+                                      [:value :conflict-evidence]
+                                      (get-in snapshots [node :conflict-evidence])))
+                          (range 3) ["n1" "n2" "n3"])
+            result (model/analyze test-map history)]
+        (is (= valid (:valid? result)) (str label ": " result)))
+      (let [empty-history (mapv #(assoc-in (snapshot-op %1 %2 [] (empty-registry) (empty-pg))
+                                         [:value :conflict-evidence] reset-evidence)
+                                (range 3) ["n1" "n2" "n3"])
+            result (model/analyze (assoc test-map :required-transport-events #{:registry-conflict-death})
+                                  empty-history)]
+        (is (false? (:valid? result)) "reset history cannot reuse conflict coverage")))
+    (when scenarios
+      (with-redefs [docker/docker!
+                    (fn [& args]
+                      (spit (last args)
+                            (if (str/includes? (second args) "conflict-evidence")
+                              (:archive (first scenarios)) "")))]
+        (let [archive (docker/retired-evidence! "n2")
+              test (assoc test-map :terminal-nodes ["n1" "n3"])
+              scenario (first scenarios)
+              history [{:type :info :f :retire :value {:lifecycle-evidence archive}}
+                       (assoc-in (snapshot-op 1 "n1" ["n1" "n3"] [] (empty-registry) (empty-pg))
+                                 [:value :conflict-evidence]
+                                 (get-in scenario [:snapshots "n1" :conflict-evidence]))
+                       (snapshot-op 2 "n3" ["n1" "n3"] [] (empty-registry) (empty-pg))]]
+          (is (= (get-in scenario [:snapshots "n2" :conflict-evidence])
+                 (:conflict-evidence archive)))
+          (is (:valid? (model/analyze test history))))))))
+
 (deftest accepts-an-exact-converged-multi-cluster-view
   (let [owners [(owner "a" [(registration nil 0 1) (registration "red" 1 2)] [])
                 (owner "b" [] [(membership nil 1 2) (membership "red" 0 3)])]
@@ -210,11 +338,8 @@
            (:missing-transport-events result)))))
 
 (deftest accepts-the-transport-event-names-emitted-by-the-live-nodes
-  (let [events {:delta-batch 1
-                :snapshot-chunk 2
-                :multi-chunk-snapshot 1
-                :registry-conflict-death 1}
-        history [(assoc-in (snapshot-op 1 "n1" [] (empty-registry) (empty-pg))
+  (let [events (assoc applied-repair-events :registry-conflict-death 1)
+        history [(assoc-in (with-conflict-evidence (snapshot-op 1 "n1" [] (empty-registry) (empty-pg)))
                            [:value :transport-events]
                            events)
                  (snapshot-op 2 "n2" [] (empty-registry) (empty-pg))
@@ -226,11 +351,9 @@
     (is (empty? (:missing-transport-events result)))))
 
 (deftest rejects-a-profile-which-never-repairs-a-multi-record-delta-run
-  (let [single-record-events {:delta-batch 3
-                              :snapshot-chunk 2
-                              :multi-chunk-snapshot 1
-                              :registry-conflict-death 1
-                              :delta-run-records-peak 1}
+  (let [single-record-events (assoc applied-repair-events
+                                   :registry-conflict-death 1
+                                   :applied-delta-run-records-peak 1)
         with-events #(assoc-in % [:value :transport-events] single-record-events)
         history [(with-events (snapshot-op 1 "n1" [] (empty-registry) (empty-pg)))
                  (with-events (snapshot-op 2 "n2" [] (empty-registry) (empty-pg)))
@@ -241,11 +364,7 @@
     (is (= 2 (:min-delta-run-records result)))))
 
 (deftest accepts-a-profile-which-repairs-a-multi-record-delta-run
-  (let [events {:delta-batch 1
-                :snapshot-chunk 1
-                :multi-chunk-snapshot 1
-                :registry-conflict-death 1
-                :delta-run-records-peak 8}
+  (let [events (assoc applied-repair-events :registry-conflict-death 1)
         history [(assoc-in (snapshot-op 1 "n1" [] (empty-registry) (empty-pg))
                            [:value :transport-events]
                            events)
@@ -253,7 +372,34 @@
                  (snapshot-op 3 "n3" [] (empty-registry) (empty-pg))]
         result (model/analyze (assoc test-map :min-delta-run-records 2) history)]
     (is (:valid? result))
-    (is (= 8 (:delta-run-records-peak result)))))
+    (is (= 2 (:delta-run-records-peak result)))))
+
+(deftest sender-attempts-never-certify-receiver-repair
+  (let [events {:delta-batch 100
+                :snapshot-chunk 100
+                :snapshot-commit 100
+                :multi-chunk-snapshot 100
+                :delta-run-records-peak 100
+                :attempted-delta-run-records-peak 100
+                :attempted-snapshot-chunks-peak 100
+                :transport-ok 100
+                :logical-drop 100
+                :registry-conflict-death 1}
+        history [(assoc-in (snapshot-op 1 "n1" [] (empty-registry) (empty-pg))
+                           [:value :transport-events] events)
+                 (snapshot-op 2 "n2" [] (empty-registry) (empty-pg))
+                 (snapshot-op 3 "n3" [] (empty-registry) (empty-pg))]
+        result (model/analyze
+                 (-> test-map
+                     (dissoc :required-transport-events)
+                     (assoc :min-delta-run-records 2))
+                 history)]
+    (is (false? (:valid? result)))
+    (is (zero? (:delta-run-records-peak result)))
+    (is (= #{:applied-delta-run-records-peak
+             :multi-chunk-snapshot-committed
+             :registry-conflict-death}
+           (:missing-transport-events result)))))
 
 (deftest rejects-internal-corruption-or-leftover-snapshot-staging
   (let [bad (-> (snapshot-op 1 "n1" [] (empty-registry) (empty-pg))
diff --git a/test/jepsen/test/group/jepsen/retired_evidence_test.clj b/test/jepsen/test/group/jepsen/retired_evidence_test.clj
index 3491e73..5f738c5 100644
--- a/test/jepsen/test/group/jepsen/retired_evidence_test.clj
+++ b/test/jepsen/test/group/jepsen/retired_evidence_test.clj
@@ -1,6 +1,8 @@
 (ns group.jepsen.retired-evidence-test
   (:require [clojure.test :refer :all]
             [group.jepsen.docker :as docker]
+            [group.jepsen.client :as client]
+            [group.jepsen.db :as group-db]
             [group.jepsen.nemesis :as group-nemesis]
             [jepsen.db :as db]
             [jepsen.nemesis :as nemesis]))
@@ -11,15 +13,18 @@
                   (fn [& args]
                     (swap! calls conj args)
                     (is (= 10000 docker/*command-timeout-ms*))
-                    (spit (last args) "n1/boot/owner/1\t:boom\n"))]
-      (is (= {:node "n1" :unexpected-deaths [{:token "n1/boot/owner/1" :reason ":boom"}]}
+                    (spit (last args) "n1/boot/owner/1\t:boom\n"))
+                  docker/decode-conflict-evidence! (constantly [])]
+      (is (= {:node "n1" :unexpected-deaths [{:token "n1/boot/owner/1" :reason ":boom"}]
+              :conflict-evidence []}
              (docker/retired-evidence! "n1")))
       (is (= ["cp" "group-jepsen-n1:/tmp/group-jepsen-unexpected-deaths"]
              (vec (take 2 (first @calls))))))))
 
 (deftest collector-distinguishes-empty-evidence-from-loss
   (doseq [contents ["" "bad\n" "token\t:boom"]]
-    (with-redefs [docker/docker! (fn [& args] (spit (last args) contents))]
+    (with-redefs [docker/docker! (fn [& args] (spit (last args) contents))
+                  docker/decode-conflict-evidence! (constantly [])]
       (if (= "" contents)
         (is (= [] (:unexpected-deaths (docker/retired-evidence! "n1"))))
         (is (thrown? Exception (docker/retired-evidence! "n1"))))))
@@ -33,6 +38,32 @@
                   (is (re-find #": > /tmp/group-jepsen-unexpected-deaths" script)))]
     (docker/reset-oracle! "n1")))
 
+(deftest workload-reset-clears-the-running-conflict-recorder
+  (let [calls (atom [])]
+    (with-redefs [docker/heal! (fn [_])
+                  docker/restart! #(swap! calls conj [:restart %])
+                  docker/reset-oracle! #(swap! calls conj [:disk-reset %])
+                  client/wait-listening! #(swap! calls conj [:listening %])
+                  client/request! (fn [node fields]
+                                    (swap! calls conj [node fields])
+                                    {:status :ok})
+                  client/wait-ready! (fn [node _] (swap! calls conj [:ready node]))]
+      (dotimes [_ 2]
+        (db/setup! (group-db/db) {:nodes ["n1"]} "n1"))
+      (is (= (vec (mapcat identity (repeat 2 [[:restart "n1"] [:disk-reset "n1"]
+                                             [:listening "n1"]
+                                             ["n1" ["reset-conflict-evidence"]]
+                                             [:ready "n1"]])))
+             @calls)))))
+
+(deftest conflict-archive-decode-fails-closed
+  (doseq [contents ["not-base64\n" "truncated"]]
+    (with-redefs [docker/docker!
+                  (fn [& args]
+                    (spit (last args)
+                          (if (.contains (second args) "conflict-evidence") contents "")))]
+      (is (thrown? Exception (docker/retired-evidence! "n1"))))))
+
 (deftest retirement-captures-after-stop-even-if-node-was-already-unavailable
   (doseq [running? [true false]]
     (let [calls (atom [])
diff --git a/test/jepsen_repair_coverage_test.exs b/test/jepsen_repair_coverage_test.exs
new file mode 100644
index 0000000..52d277e
--- /dev/null
+++ b/test/jepsen_repair_coverage_test.exs
@@ -0,0 +1,224 @@
+Code.require_file("jepsen/repair_coverage.exs", __DIR__)
+
+defmodule Group.JepsenRepairCoverageTest do
+  use ExUnit.Case, async: false
+
+  alias Group.TestCluster, as: Cluster
+  alias Group.JepsenRepairProbe, as: Probe
+  alias Group.Jepsen.RepairCoverage, as: Coverage
+  alias Group.Jepsen.Transport.Stats
+  alias Group.Replica.Data
+
+  @moduletag :capture_log
+  @moduletag tmp_dir: "repair_coverage_#{System.pid()}"
+  @moduletag timeout: 120_000
+
+  test "restart truncates an interrupted append before accepting new evidence", %{tmp_dir: dir} do
+    path = Path.join(dir, "interrupted")
+    committed = "multi_chunk_snapshot_committed\t1\n"
+
+    for suffix <- ["applied_delta_run_records_peak\t", "applied_delta_run_records_peak\t99"] do
+      File.write!(path, committed <> suffix)
+      start_supervised!({Coverage, path: path})
+      assert Coverage.snapshot() == %{multi_chunk_snapshot_committed: 1}
+      assert File.read!(path) == committed
+
+      Coverage.observe(:delta, 0, 2)
+      expected = %{multi_chunk_snapshot_committed: 1, applied_delta_run_records_peak: 2}
+      assert Coverage.snapshot() == expected
+      assert File.read!(path) == committed <> "applied_delta_run_records_peak\t2\n"
+
+      stop_supervised!(Coverage)
+      start_supervised!({Coverage, path: path})
+      assert Coverage.snapshot() == expected
+      stop_supervised!(Coverage)
+    end
+  end
+
+  test "malformed committed records fail without truncating the log", %{tmp_dir: dir} do
+    path = Path.join(dir, "malformed")
+    contents = "applied_delta_run_records_peak\t2\tunexpected\npartial"
+    File.write!(path, contents)
+    assert {:error, {reason, stack}} = GenServer.start(Coverage, path: path)
+    assert %MatchError{} = Exception.normalize(:error, reason, stack)
+    assert File.read!(path) == contents
+  end
+
+  for transport <- [
+        Group.Jepsen.Transport.Distribution,
+        Group.Jepsen.Transport.TCP,
+        Group.Jepsen.Transport.Chaos
+      ] do
+    test "#{transport}: only receiver commits certify repair", %{tmp_dir: dir} do
+      transport = unquote(transport)
+      peers = Cluster.start_peers(2, schedulers: 2)
+      on_exit(fn -> Cluster.stop_peers(peers) end)
+      [{_, source}, {_, receiver}] = peers
+
+      for {node, suffix} <- [{source, "source"}, {receiver, "receiver"}] do
+        :ok = rpc(node, Probe, :boot, [Path.join(dir, suffix)])
+      end
+
+      name = :coverage_probe
+
+      for node <- [source, receiver] do
+        {:ok, _} =
+          Cluster.start_group(node,
+            name: name,
+            shards: 1,
+            log: false,
+            replica_transport: transport,
+            replicated_anti_entropy_interval: 60_000,
+            replicated_peer_lease_timeout: 120_000,
+            replicated_sender_buffer_size: 1
+          )
+      end
+
+      Cluster.assert_eventually(fn ->
+        receiver in rpc(source, Group, :nodes, [name]) and
+          source in rpc(receiver, Group, :nodes, [name])
+      end)
+
+      rpc(source, Stats, :block, [receiver])
+      rpc(receiver, Stats, :block, [source])
+      pid = Cluster.spawn_join(source, name, "one", %{})
+      Cluster.flush_shards(source, name)
+      stream = rpc(source, Data, :local_stream_id, [name, 0, nil])
+      version = Group.Replica.WireProtocol.version()
+      mutation = fn key -> {:join, nil, key, pid, %{}, 1, :normal, source} end
+      records = [{1, [mutation.("one")]}, {2, [mutation.("two")]}]
+      delta = {:delta_batch, version, [{stream, 1, records, 2}]}
+      chunk1 = {:snapshot_chunk, version, stream, 4, 1, [], [{"one", pid, %{}, 1}]}
+      chunk2 = {:snapshot_chunk, version, stream, 4, 2, [], [{"two", pid, %{}, 1}]}
+      commit = {:snapshot_commit, version, stream, 4, 2, 0, 2}
+
+      # Every live wrapper's logical gate reports attempts without delegating.
+      for message <- [delta, chunk1, chunk2, commit] do
+        assert :ok = rpc(source, transport, :outgoing, [name, receiver, 0, message, []])
+      end
+
+      assert events(receiver, name) == %{}
+      assert rpc(source, Coverage, :snapshot, []) == %{}
+      attempted = rpc(source, Stats, :snapshot, [])
+      assert attempted.attempted_delta_run_records_peak == 2
+      assert attempted.attempted_snapshot_chunks_peak == 2
+
+      # Bypass the logical gate, but still use the actual selected data lane.
+      rpc(source, Stats, :unblock, [receiver])
+      stale_generation = Group.Replica.WireProtocol.new_generation()
+      stale = stream |> put_elem(2, stale_generation) |> put_elem(5, stale_generation)
+      assert Group.Replica.WireProtocol.valid_stream_id?(stale)
+      rejected = {:delta_batch, version, [{stale, 1, records, 2}]}
+      rejected_snapshot = Enum.map([chunk1, chunk2, commit], &put_elem(&1, 2, stale))
+      send_frames(source, receiver, name, transport, [rejected | rejected_snapshot])
+      assert events(receiver, name) == %{}
+
+      # A gap and a rejected mutation cannot inflate the actual contiguous run.
+      gap = {:delta_batch, version, [{stream, 2, [List.last(records)], 2}]}
+      invalid = {:delta_batch, version, [{stream, 1, [{1, [:invalid]}], 1}]}
+      send_frames(source, receiver, name, transport, [gap, invalid])
+      assert events(receiver, name) == %{}
+
+      # One accepted record followed by duplicate-only traffic is not a run of two.
+      single = {:delta_batch, version, [{stream, 1, [hd(records)], 1}]}
+
+      deliver_until(source, receiver, name, transport, [single], fn ->
+        rpc(receiver, Data, :replica_cursor, [name, 0, stream]) == 1
+      end)
+
+      send_frames(source, receiver, name, transport, [single, single])
+      assert events(receiver, name) == %{applied_delta_run_records_peak: 1}
+
+      # Two new records in one run really advance two, excluding the duplicate prefix.
+      run = {:delta_batch, version, [{stream, 1, records ++ [{3, [mutation.("three")]}], 3}]}
+
+      deliver_until(source, receiver, name, transport, [run], fn ->
+        rpc(receiver, Data, :replica_cursor, [name, 0, stream]) == 3
+      end)
+
+      assert events(receiver, name) == %{applied_delta_run_records_peak: 2}
+      assert [{^pid, %{}}] = rpc(receiver, Group, :members, [name, "three"])
+
+      # All chunks without a terminal manifest are still uncommitted.
+      send_frames(source, receiver, name, transport, [chunk1, chunk2, chunk1])
+      assert rpc(receiver, Data, :replica_cursor, [name, 0, stream]) == 3
+      assert events(receiver, name) == %{applied_delta_run_records_peak: 2}
+
+      # Supersede the provisional transfer. Commit alone, incomplete assembly,
+      # duplicate chunks, and a stale terminal frame preserve the old slice.
+      chunk1 = put_elem(chunk1, 3, 5)
+      chunk2 = put_elem(chunk2, 3, 5)
+      commit = put_elem(commit, 3, 5)
+      send_frames(source, receiver, name, transport, [commit, chunk1, chunk1])
+      assert rpc(receiver, Data, :replica_cursor, [name, 0, stream]) == 3
+      assert events(receiver, name) == %{applied_delta_run_records_peak: 2}
+      send_frames(source, receiver, name, transport, [put_elem(commit, 2, stale)])
+      assert events(receiver, name) == %{applied_delta_run_records_peak: 2}
+
+      deliver_until(source, receiver, name, transport, [chunk2, chunk1, commit], fn ->
+        rpc(receiver, Data, :replica_cursor, [name, 0, stream]) == 5
+      end)
+
+      assert rpc(receiver, Group, :members, [name, "three"]) == []
+      assert [{^pid, %{}}] = rpc(receiver, Group, :members, [name, "two"])
+
+      assert rpc(receiver, :sys, :get_state, [Group.Replica.shard_name(name, 0)]).snapshot_transfers ==
+               %{}
+
+      evidence = events(receiver, name)
+
+      emitted =
+        rpc(receiver, Stats, :snapshot, [])
+        |> Map.take([:applied_delta_run_records_peak, :multi_chunk_snapshot_committed])
+
+      assert emitted == evidence
+      encoded = rpc(receiver, Group.Jepsen.EDN, :encode, [emitted])
+      assert encoded == String.trim(File.read!("test/jepsen/fixtures/applied-repair-events.edn"))
+      send_frames(source, receiver, name, transport, [run, chunk1, chunk2, commit])
+      assert events(receiver, name) == evidence
+
+      # The durable emitted evidence, not transient sender stats, survives restart.
+      rpc(receiver, GenServer, :stop, [Coverage])
+      rpc(receiver, Probe, :start_coverage, [Path.join(dir, "receiver")])
+      assert rpc(receiver, Coverage, :snapshot, []) == evidence
+
+      # A genuinely new BEAM loads the same completed evidence. Keep at most two
+      # peers alive, and never recompile an instrumented module under a live Group.
+      Cluster.stop_peers(peers)
+      replacement = Cluster.start_peers(1, schedulers: 2)
+      on_exit(fn -> Cluster.stop_peers(replacement) end)
+      [{_, restarted}] = replacement
+      rpc(restarted, Probe, :boot, [Path.join(dir, "receiver")])
+      assert rpc(restarted, Coverage, :snapshot, []) == evidence
+
+      # A new history clears persisted evidence even if the VM is already up.
+      File.rm!(Path.join(dir, "receiver"))
+      assert rpc(restarted, Coverage, :snapshot, []) == %{}
+    end
+  end
+
+  defp rpc(node, module, function, args), do: :erpc.call(node, module, function, args)
+
+  defp events(node, name) do
+    rpc(node, Probe, :barrier, [name])
+    rpc(node, Coverage, :snapshot, [])
+  end
+
+  defp send_frames(source, receiver, name, transport, frames) do
+    for frame <- frames do
+      rpc(source, transport, :outgoing, [name, receiver, 0, frame, []])
+    end
+
+    # TCP/outbox and chaos forwarding is asynchronous. This bound is only for
+    # negative observations; positive assertions below always await exact state.
+    Process.sleep(150)
+    rpc(receiver, Probe, :barrier, [name])
+  end
+
+  defp deliver_until(source, receiver, name, transport, frames, predicate) do
+    Cluster.assert_eventually(fn ->
+      send_frames(source, receiver, name, transport, frames)
+      predicate.()
+    end)
+  end
+end
diff --git a/test/support/jepsen_repair_probe.ex b/test/support/jepsen_repair_probe.ex
new file mode 100644
index 0000000..8526636
--- /dev/null
+++ b/test/support/jepsen_repair_probe.ex
@@ -0,0 +1,27 @@
+defmodule Group.JepsenRepairProbe do
+  @moduledoc false
+
+  def boot(path) do
+    System.put_env("GROUP_JEPSEN_LIBRARY_ONLY", "1")
+    Code.require_file("../jepsen/node.exs", __DIR__)
+    :ok = apply(Group.Jepsen.RepairCoverage, :install!, [])
+    start_coverage(path)
+  end
+
+  def start_coverage(path) do
+    {:ok, pid} = apply(Group.Jepsen.RepairCoverage, :start_link, [[path: path]])
+    Process.unlink(pid)
+    :ok
+  end
+
+  def barrier(name) do
+    :sys.get_state(Group.Replica.shard_name(name, 0))
+    # Calls from the shard order its earlier coverage casts ahead of this reply.
+    :sys.replace_state(Group.Replica.shard_name(name, 0), fn state ->
+      apply(Group.Jepsen.RepairCoverage, :snapshot, [])
+      state
+    end)
+
+    :ok
+  end
+end