diff --git a/.formatter.exs b/.formatter.exs
index d2cda26..a2abb7d 100644
--- a/.formatter.exs
+++ b/.formatter.exs
@@ -1,4 +1,8 @@
 # Used by "mix format"
 [
-  inputs: ["{mix,.formatter}.exs", "{config,lib,test}/**/*.{ex,exs}"]
+  inputs: [
+    "{mix,.formatter}.exs",
+    "{config,lib,test}/**/*.{ex,exs}",
+    "priv/bench/comparison.exs"
+  ]
 ]
diff --git a/.github/workflows/campaigns.yml b/.github/workflows/campaigns.yml
new file mode 100644
index 0000000..4d6310a
--- /dev/null
+++ b/.github/workflows/campaigns.yml
@@ -0,0 +1,148 @@
+name: Fault campaigns
+
+on:
+  schedule:
+    - cron: "23 3 * * *"
+  workflow_dispatch:
+  release:
+    types: [published]
+
+permissions:
+  contents: read
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: false
+
+defaults:
+  run:
+    shell: bash
+
+env:
+  MIX_ENV: test
+  ERL_FLAGS: "+S 4:4"
+  GROUP_HISTORY_STEPS: "2000"
+  GROUP_TEST_DIAGNOSTICS: ${{ github.workspace }}/_build/test-diagnostics
+  ERL_CRASH_DUMP: ${{ github.workspace }}/_build/test-diagnostics/erl_crash.dump
+
+jobs:
+  histories-and-faults:
+    name: Seed ${{ matrix.seed }} / ${{ matrix.schedulers }} peer schedulers
+    runs-on: ubuntu-24.04
+    timeout-minutes: 30
+    strategy:
+      fail-fast: false
+      matrix:
+        seed: [104729, 130363, 155921]
+        schedulers: ["1", "2", "4"]
+    env:
+      GROUP_PEER_SCHEDULERS: ${{ matrix.schedulers }}
+      CAMPAIGN_SEED: ${{ matrix.seed }}
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+      - uses: erlef/setup-beam@54075bcc5e249e4758d363f27d099f55d843f124 # v1.24.1
+        with:
+          elixir-version: "> 0"
+          otp-version: "> 0"
+      - uses: actions/setup-java@de7274f081f381c8f8158605e0321c36c376e2e6 # v6.0.1
+        with:
+          distribution: temurin
+          java-version: "21"
+      - run: mix deps.get
+      - name: Repeat histories, partitions, churn, restarts and race regressions
+        run: |
+          mkdir -p "$GROUP_TEST_DIAGNOSTICS"
+          mix compile --warnings-as-errors
+          for iteration in 0 1 2 3 4; do
+            seed=$((CAMPAIGN_SEED + iteration))
+            mix test --warnings-as-errors --seed "$seed" 2>&1 |
+              tee "$GROUP_TEST_DIAGNOSTICS/run-$seed.log"
+          done
+      - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        if: always()
+        with:
+          name: campaign-${{ matrix.seed }}-${{ matrix.schedulers }}
+          path: _build/test-diagnostics
+          retention-days: 14
+
+  qualification:
+    name: Protocol mutations and live checker qualification
+    runs-on: ubuntu-24.04
+    timeout-minutes: 90
+    env:
+      GROUP_HISTORY_STEPS: "100"
+      GROUP_PEER_SCHEDULERS: "2"
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+      - uses: erlef/setup-beam@54075bcc5e249e4758d363f27d099f55d843f124 # v1.24.1
+        with:
+          elixir-version: "> 0"
+          otp-version: "> 0"
+      - uses: actions/setup-java@de7274f081f381c8f8158605e0321c36c376e2e6 # v6.0.1
+        with:
+          distribution: temurin
+          java-version: "21"
+      - run: mix deps.get
+      - name: Run the upstream PR gate before qualification
+        run: |
+          mkdir -p "$GROUP_TEST_DIAGNOSTICS"
+          mix test --warnings-as-errors 2>&1 | tee "$GROUP_TEST_DIAGNOSTICS/baseline.log"
+      - name: Require protocol mutants and injected checker faults to be detected
+        run: |
+          GROUP_JEPSEN_SKIP_CHECKER=1 test/jepsen/qualify.sh 2>&1 |
+            tee "$GROUP_TEST_DIAGNOSTICS/qualification.log"
+      - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        if: always()
+        with:
+          name: qualification-reports
+          path: |
+            _build/test-diagnostics
+            tmp/mutation/**/*.log
+            test/jepsen/.cache/qualification.*
+            test/jepsen/store
+          include-hidden-files: true
+          retention-days: 14
+
+  soak:
+    name: Jepsen ${{ matrix.transport }} / ${{ matrix.scenario }}
+    needs: qualification
+    runs-on: ubuntu-24.04
+    timeout-minutes: 180
+    strategy:
+      fail-fast: false
+      matrix:
+        transport: [distribution, tcp, chaos]
+        scenario: [mixed, permanent]
+    env:
+      GROUP_JEPSEN_SKIP_CHECKER: "1"
+      GROUP_JEPSEN_CAMPAIGN_TRANSPORT: ${{ matrix.transport }}
+      GROUP_JEPSEN_CAMPAIGN_SCENARIO: ${{ matrix.scenario }}
+      GROUP_JEPSEN_CAMPAIGN_ARTIFACT_DIR: ${{ github.workspace }}/_build/jepsen-campaign
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+      - uses: erlef/setup-beam@54075bcc5e249e4758d363f27d099f55d843f124 # v1.24.1
+        with:
+          elixir-version: "> 0"
+          otp-version: "> 0"
+      - uses: actions/setup-java@de7274f081f381c8f8158605e0321c36c376e2e6 # v6.0.1
+        with:
+          distribution: temurin
+          java-version: "21"
+      - run: mix deps.get
+      # Split the upstream six-profile gate across jobs without reducing its
+      # default 20 five-minute histories per profile.
+      - run: test/jepsen/campaign.sh
+      - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        if: always()
+        with:
+          name: soak-${{ matrix.transport }}-${{ matrix.scenario }}
+          path: |
+            _build/jepsen-campaign
+            test/jepsen/store
+          retention-days: 14
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
new file mode 100644
index 0000000..bba6b51
--- /dev/null
+++ b/.github/workflows/ci.yml
@@ -0,0 +1,114 @@
+name: CI
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+  release:
+    types: [published]
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
+defaults:
+  run:
+    shell: bash
+
+env:
+  MIX_ENV: test
+  ERL_FLAGS: "+S 4:4"
+  GROUP_PEER_SCHEDULERS: "2"
+  GROUP_TEST_DIAGNOSTICS: ${{ github.workspace }}/_build/test-diagnostics
+  ERL_CRASH_DUMP: ${{ github.workspace }}/_build/test-diagnostics/erl_crash.dump
+
+jobs:
+  local:
+    name: Local checks
+    runs-on: ubuntu-24.04
+    timeout-minutes: 5
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+      - uses: erlef/setup-beam@54075bcc5e249e4758d363f27d099f55d843f124 # v1.24.1
+        with:
+          elixir-version: "> 0"
+          otp-version: "> 0"
+      - uses: actions/setup-java@de7274f081f381c8f8158605e0321c36c376e2e6 # v6.0.1
+        with:
+          distribution: temurin
+          java-version: "21"
+      - run: mix deps.get
+      - name: Formatting, compiler checks, local examples and histories
+        run: |
+          mkdir -p "$GROUP_TEST_DIAGNOSTICS"
+          mix format --check-formatted 2>&1 | tee "$GROUP_TEST_DIAGNOSTICS/format.log"
+          mix compile --warnings-as-errors 2>&1 | tee "$GROUP_TEST_DIAGNOSTICS/compile.log"
+          mix test --only local --warnings-as-errors 2>&1 | tee "$GROUP_TEST_DIAGNOSTICS/local.log"
+      - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        if: failure()
+        with:
+          name: local-failure
+          path: _build/test-diagnostics
+          retention-days: 14
+
+  test:
+    name: ${{ matrix.latest && 'Latest stable Elixir / OTP' || format('Elixir {0} / OTP {1}', matrix.elixir, matrix.otp) }}
+    needs: local
+    runs-on: ubuntu-24.04
+    timeout-minutes: 15
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - elixir: "1.19"
+            otp: "26"
+          - elixir: "1.19"
+            otp: "27"
+          - elixir: "1.19"
+            otp: "28"
+          - elixir: "1.20"
+            otp: "27"
+          - elixir: "1.20"
+            otp: "28"
+          # Unlike "latest", these ranges exclude prereleases.
+          - elixir: "> 0"
+            otp: "> 0"
+            latest: true
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+
+      - uses: erlef/setup-beam@54075bcc5e249e4758d363f27d099f55d843f124 # v1.24.1
+        with:
+          elixir-version: ${{ matrix.elixir }}
+          otp-version: ${{ matrix.otp }}
+
+      - uses: actions/setup-java@de7274f081f381c8f8158605e0321c36c376e2e6 # v6.0.1
+        with:
+          distribution: temurin
+          java-version: "21"
+
+      - name: Install dependencies
+        run: mix deps.get
+
+      - name: Compile with warnings as errors
+        run: |
+          mkdir -p "$GROUP_TEST_DIAGNOSTICS"
+          mix compile --warnings-as-errors 2>&1 | tee "$GROUP_TEST_DIAGNOSTICS/compile.log"
+
+      - name: Run local and distributed tests
+        run: mix test --warnings-as-errors 2>&1 | tee "$GROUP_TEST_DIAGNOSTICS/test.log"
+
+      - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        if: failure()
+        with:
+          name: runtime-failure-${{ strategy.job-index }}
+          path: _build/test-diagnostics
+          retention-days: 14
diff --git a/.github/workflows/performance.yml b/.github/workflows/performance.yml
new file mode 100644
index 0000000..5b98d1e
--- /dev/null
+++ b/.github/workflows/performance.yml
@@ -0,0 +1,83 @@
+name: Performance comparison
+
+on:
+  workflow_dispatch:
+    inputs:
+      baseline:
+        description: "Baseline ref (use a previous release tag for release review)"
+        type: string
+        default: main
+        required: true
+  release:
+    types: [published]
+
+permissions:
+  contents: read
+
+defaults:
+  run:
+    shell: bash
+
+jobs:
+  compare:
+    runs-on: ubuntu-24.04
+    timeout-minutes: 60
+    env:
+      ERL_FLAGS: "+S 4:4"
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          path: candidate
+          fetch-depth: 0
+          persist-credentials: false
+      - name: Select baseline
+        id: baseline
+        env:
+          BASELINE: ${{ inputs.baseline }}
+        run: |
+          if [ -z "$BASELINE" ]; then
+            BASELINE=$(git -C candidate describe --tags --abbrev=0 "${GITHUB_SHA}^")
+          fi
+          printf 'ref=%s\n' "$BASELINE" >> "$GITHUB_OUTPUT"
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          path: baseline
+          ref: ${{ steps.baseline.outputs.ref }}
+          persist-credentials: false
+      - uses: erlef/setup-beam@54075bcc5e249e4758d363f27d099f55d843f124 # v1.24.1
+        with:
+          elixir-version: "> 0"
+          otp-version: "> 0"
+      - name: Record environment and revisions
+        run: |
+          mkdir -p reports
+          {
+            elixir --version
+            uname -a
+            lscpu
+            git -C baseline rev-parse HEAD
+            git -C candidate rev-parse HEAD
+          } | tee reports/environment.log
+      - name: Benchee baseline and candidate on the same runner
+        run: |
+          GROUP_BENCH_ROOT="$PWD/baseline" \
+          GROUP_BENCH_OUTPUT="$PWD/reports/baseline.benchee" \
+          GROUP_BENCH_TAG=baseline \
+            elixir candidate/priv/bench/comparison.exs 2>&1 | tee reports/benchee-baseline.log
+          GROUP_BENCH_ROOT="$PWD/candidate" \
+          GROUP_BENCH_OUTPUT="$PWD/reports/candidate.benchee" \
+          GROUP_BENCH_BASELINE="$PWD/reports/baseline.benchee" \
+            elixir candidate/priv/bench/comparison.exs 2>&1 | tee reports/benchee-candidate.log
+      - name: Distributed load and recovery, baseline then candidate
+        run: |
+          epmd -daemon
+          for revision in baseline candidate; do
+            timeout 20m bash "$revision/priv/bench/run_distributed.sh" 2>&1 |
+              tee "reports/distributed-$revision.log"
+          done
+      - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        if: always()
+        with:
+          name: performance-comparison
+          path: reports
+          retention-days: 30
diff --git a/README.md b/README.md
index 66f4536..2679d59 100644
--- a/README.md
+++ b/README.md
@@ -619,6 +619,28 @@ See [`test/README.md`](test/README.md) for the every-PR gate, shrinkable
 StreamData lifecycle-model tests, bounded TLA+ models, and the nightly
 three-node Jepsen transport/lifecycle campaign.
 
+GitHub Actions runs fast local checks first, then the full suite on pull requests,
+pushes to `main`, and releases using Elixir 1.19 / OTP 26–28, Elixir 1.20 / OTP 27–28, and the latest
+stable Elixir / OTP pair. Version ranges pick up new patch releases automatically;
+the latest-stable job also picks up new minor and major releases, excluding
+prereleases. Every job treats compilation and test warnings as errors. Nightly
+campaigns vary seeds and peer scheduler counts, run larger generated histories,
+and check targeted mutations. Performance comparisons run separately.
+See [`test/README.md`](test/README.md#ci-lanes) for lane budgets and release review.
+
+To run the same checks locally:
+
+```bash
+export MIX_ENV=test
+export ERL_FLAGS="+S 4:4"
+export GROUP_PEER_SCHEDULERS=2
+mix deps.get
+mix format --check-formatted
+mix compile --warnings-as-errors
+mix test --only local --warnings-as-errors # no distribution startup
+mix test --warnings-as-errors
+```
+
 ## Benchmarks
 
 ```bash
diff --git a/priv/bench/comparison.exs b/priv/bench/comparison.exs
new file mode 100644
index 0000000..cfc22f8
--- /dev/null
+++ b/priv/bench/comparison.exs
@@ -0,0 +1,56 @@
+# Use the same harness and Benchee version for both revisions. Only the Group
+# path changes; this also works when the baseline predates this harness.
+root = System.get_env("GROUP_BENCH_ROOT") || Path.expand("../..", __DIR__)
+Mix.install([{:group, path: root}, {:benchee, "== 1.5.1"}])
+
+{:ok, supervisor} = Group.start_link(name: :comparison, shards: 8, log: false)
+parent = self()
+
+members =
+  for _ <- 1..128 do
+    spawn_link(fn ->
+      :ok = Group.join(:comparison, "members", %{})
+      send(parent, {:ready, self()})
+
+      receive do
+        :stop -> :ok
+      end
+    end)
+  end
+
+for pid <- members do
+  receive do
+    {:ready, ^pid} -> :ok
+  after
+    5_000 -> raise "benchmark member did not become ready"
+  end
+end
+
+:ok = Group.register(:comparison, "lookup", %{value: 1})
+output = System.get_env("GROUP_BENCH_OUTPUT", "comparison.benchee")
+baseline = System.get_env("GROUP_BENCH_BASELINE")
+
+try do
+  Benchee.run(
+    %{
+      "lookup" => fn -> Group.lookup(:comparison, "lookup") end,
+      "members/128" => fn -> Group.members(:comparison, "members") end,
+      "register/unregister" => fn ->
+        :ok = Group.register(:comparison, "registry-cycle", %{})
+        :ok = Group.unregister(:comparison, "registry-cycle")
+      end,
+      "join/leave" => fn ->
+        :ok = Group.join(:comparison, "pg-cycle", %{})
+        :ok = Group.leave(:comparison, "pg-cycle")
+      end
+    },
+    time: 3,
+    warmup: 1,
+    memory_time: 1,
+    save: [path: output, tag: System.get_env("GROUP_BENCH_TAG", "candidate")],
+    load: if(baseline, do: [baseline], else: [])
+  )
+after
+  for pid <- members, do: send(pid, :stop)
+  Supervisor.stop(supervisor)
+end
diff --git a/test/README.md b/test/README.md
index e0a9119..82d7cbb 100644
--- a/test/README.md
+++ b/test/README.md
@@ -5,7 +5,7 @@
 ```bash
 mix test                           # every-PR ExUnit/property/chaos/checker gate
 mix test.soak                      # nightly six-profile Jepsen campaign
-mix test test/group_test.exs       # local only
+mix test --only local              # all local examples, histories and regressions
 mix test test/distributed_test.exs # distributed only
 mix test test/replica_adversarial_test.exs # seeded transport chaos
 mix test test/replica_model_property_test.exs # shrinkable model-based histories
@@ -13,18 +13,74 @@ test/jepsen/run.sh                 # one OS-partition/restart Jepsen model test
 ```
 
 `mix test` preserves normal Mix test arguments while always running the pure
-Jepsen lifecycle-checker qualification after ExUnit. It does not require
-Docker. `mix test.soak` first runs that complete PR gate, kills every defined
+Jepsen lifecycle-checker qualification after ExUnit. It requires Java 21 or newer
+but not Docker. `mix test.soak` first runs that complete PR gate, kills every defined
 protocol mutant, runs live positive/negative checker qualification, and then
 runs the distribution/TCP/chaos × mixed/permanent Jepsen campaign. The soak
 defaults to 20 five-minute fault histories per combination and is intended for
 nightly and release qualification rather than individual edits.
 
+## CI lanes
+
+| Lane | Scope and budget |
+|------|------------------|
+| PR / `main` | Latest-stable formatting, compiler checks and local tests first (5-minute cap), then the six supported-runtime jobs with the full suite and fresh peers (15-minute cap each). |
+| Nightly, 03:23 UTC / on demand | Five seeds per job across three starting seeds and 1/2/4 peer schedulers; 2,000-operation histories on 1/4/8 shards, plus existing partitions, churn, restart and race scenarios (30-minute cap per job). |
+| Nightly / release qualification | Upstream protocol mutation and live checker qualification (90-minute cap), followed by six independent transport/scenario jobs, each running 20 five-minute Jepsen histories (180-minute cap per job). |
+| Dedicated performance | Manually selected baseline versus candidate on one runner: pinned Benchee harness and existing distributed load/recovery benchmarks. Reports retained for 30 days; no noisy hosted-runner threshold gates (60-minute cap). |
+| Release | Publishing reruns the runtime matrix, fault campaigns and performance comparison against the previous tag. **Before publishing**, manually run these workflows on the candidate ref and review the performance report. Publication is not blocked automatically. |
+
+Nightly and release workflows become available after merging onto the default
+branch. PR checks do not run benchmarks or mutation campaigns. Branch pushes do
+not duplicate the PR's runtime matrix.
+
+The histories check a sequential reference model for registry/PG metadata and
+named-cluster disconnects. They complement, rather than replace, the concurrent
+and real-peer tests. The upstream mutation and soak gates remain the source of
+truth for protocol qualification. Mixed Group-version wire
+compatibility is not established by this runtime matrix; add a dedicated
+mixed-version scenario before promising rolling upgrades across library versions.
+
+### Replay and diagnostics
+
+```bash
+GROUP_PEER_SCHEDULERS=1 GROUP_HISTORY_STEPS=2000 \
+  GROUP_TEST_DIAGNOSTICS=_build/test-diagnostics \
+  mix test --seed 104729 --warnings-as-errors
+
+# Protocol mutations run in isolated copies under tmp/mutation.
+mix run test/mutation/run.exs
+```
+
+CI failure artifacts retain output, seeds, generated histories, runtime/config
+details, peer/test identities, and snapshots taken **before peer teardown**:
+topology, ETS sizes, bounded shard state (including pending buffer counts) and
+process queue sizes. Unreachable peers and timed-out state calls are recorded as
+unavailable, rather than delaying cleanup indefinitely. PR artifacts are kept
+for 14 days on failure; campaign reports are kept on success too.
+
+### Local concurrency and synchronization
+
+`group_test.exs` contains separate subsystem modules using `Group.LocalCase`.
+Keeping the existing file entry point preserves focused commands and avoids a
+large file-movement diff; ExUnit schedules the modules independently. The
+process-group ingress, clusters, counts and event-batching modules remain serial because they change
+VM-global trace patterns. Diagnostics tests also remain serial because they
+change an environment variable. Other local tests use uniquely named Group
+instances and run asynchronously.
+
+Fairness regressions wait until requests appear in the suspended shard's mailbox
+instead of sleeping for a presumed delivery delay. Keep deliberate timeout and
+negative-event assertions distinct from synchronization waits. Destructive
+distributed scenarios always get fresh peers; there is no shared peer pool.
+
 ## Test files
 
 | File | What it tests |
 |------|---------------|
 | `group_test.exs` | Single-node: register/unregister, join/leave, materialized exact/prefix counts, restart rebuild and journal-replay crash boundaries, monitors, named clusters, and concurrent operations |
+| `history_test.exs` | Seeded local registry/PG/cluster histories against a reference model |
+| `diagnostics_test.exs` | Failure output, shard snapshots and unreachable-peer diagnostics |
 | `distributed_test.exs` | Multi-node: replication, peer discovery, node disconnect cleanup, partition healing, conflict resolution, event ordering, rolling restarts, and adversarial replica-transport loss/busy/snapshot recovery |
 | `anti_entropy_fault_regression_test.exs` | Three-node regressions for hidden-winner projection, receiver restart eviction, nodedown/lease lane retirement, authority gaps and cross-lane races, in-flight conflict fencing, crash-journal replay, cursorless/interrupted snapshot repair, malformed ingress, and sideband rediscovery |
 | `replica_adversarial_test.exs` | Reproducible three-node mixed-operation state machines: drops, busy returns, duplication, reordering, bounded delay, oplog pruning, conflicts, owner death, and named-cluster epoch churn, followed by exact convergence/dead-owner/internal-index checks |
@@ -75,19 +131,27 @@ live checker rejects injected faults. Chaos/mixed uses a larger repair window
 and is invalid unless it observes a multi-record delta run; the other profiles
 retain the one-record stress configuration.
 
-## How distribution works
+CI splits the soak matrix with `GROUP_JEPSEN_CAMPAIGN_TRANSPORT` and
+`GROUP_JEPSEN_CAMPAIGN_SCENARIO`. Omit these filters to run all six profiles;
+the history count and duration defaults are unchanged. This keeps each job
+within GitHub's runtime limit without reducing qualification coverage.
 
-The test node starts as a named Erlang node in `test_helper.exs`:
+## How distribution works
 
-```elixir
-Node.start(:"test_12345@127.0.0.1", :longnames)
-Node.set_cookie(:group_test)
-```
+`test_helper.exs` starts ExUnit only. `Group.TestCluster.start_peers/2` starts
+EPMD/distribution lazily, so `mix test --only local` does not start distribution.
+The test node and peers include the OS pid in their names to avoid cross-VM
+collisions. Peers use the test node's cookie.
 
 Peer nodes are real BEAM VMs started via OTP's `:peer` module (not
 `Node.spawn`). Each peer has its own schedulers, memory, and GC — they
 communicate over Erlang distribution just like production nodes.
 
+Correctness peers default to two schedulers, independently of the parent VM's
+`ERL_FLAGS`. Override with `GROUP_PEER_SCHEDULERS=1` (1–64) or
+`TestCluster.start_peers(3, schedulers: 1)`. Performance runs have separate
+scheduler settings.
+
 `:prevent_overlapping_partitions` is set to `false` on all nodes (test node
 and peers). Without this, disconnecting two peers from each other would also
 disconnect them from the test node, making partition tests impossible.
@@ -164,8 +228,10 @@ TestCluster.spawn_register_update_unregister(node_a, :test, "user/1", %{v: 1}, %
 
 `spawn_register` accepts `flush_shards: num_shards` which calls
 `:sys.get_state` on the target shard's GenServer after registration. This
-blocks until all pending messages (nodedown, replicate, etc.) are processed
-on that shard — useful in partition tests where you need to guarantee ordering.
+synchronizes with that shard after the write; it is not a pre-write barrier
+and does not flush buffered replication. Use `TestCluster.flush_shards/2` to
+flush replication and `TestCluster.assert_group_nodes/3` to wait for peer
+discovery or nodedown cleanup on every shard.
 
 ```elixir
 TestCluster.spawn_register(node_a, :test, "key", %{}, flush_shards: 4)
@@ -196,6 +262,9 @@ TestCluster.monitor_nodes_on(node_a, self())
 assert_receive {:nodedown_on_remote, ^node_b}, 5000
 ```
 
+`monitor_nodes_on/2` returns only after the remote monitor is installed.
+Receiving its notification does not mean Group's shards have processed nodedown.
+
 #### Network partitions
 
 ```elixir
@@ -208,6 +277,24 @@ partitions don't work reliably because the test node bridges them — Erlang
 distribution is fully meshed, so if the test node can reach both peers, they
 can reach each other through it.
 
+Before disconnecting, wait for discovery on **every shard**, not just
+`Node.list/0` connectivity. In-flight discovery sends can otherwise trigger
+reconnect retries. After disconnecting, wait for every shard's peer state
+before writing partition-local data:
+
+```elixir
+TestCluster.assert_group_nodes(node_a, :test, [node_b, node_c])
+TestCluster.assert_group_nodes(node_b, :test, [node_a, node_c])
+TestCluster.assert_group_nodes(node_c, :test, [node_a, node_b])
+
+TestCluster.disconnect_nodes(node_c, node_a)
+TestCluster.disconnect_nodes(node_c, node_b)
+
+TestCluster.assert_group_nodes(node_a, :test, [node_b])
+TestCluster.assert_group_nodes(node_b, :test, [node_a])
+TestCluster.assert_group_nodes(node_c, :test, [])
+```
+
 #### Polling for eventual consistency
 
 ```elixir
diff --git a/test/diagnostics_test.exs b/test/diagnostics_test.exs
new file mode 100644
index 0000000..e3b4354
--- /dev/null
+++ b/test/diagnostics_test.exs
@@ -0,0 +1,61 @@
+defmodule Group.DiagnosticsTest do
+  # These checks temporarily change a process-wide environment variable.
+  use ExUnit.Case, async: false
+
+  @moduletag :local
+  @moduletag :tmp_dir
+  @moduletag :capture_log
+
+  setup %{tmp_dir: directory} do
+    previous = System.get_env("GROUP_TEST_DIAGNOSTICS")
+    System.put_env("GROUP_TEST_DIAGNOSTICS", directory)
+
+    on_exit(fn ->
+      if previous do
+        System.put_env("GROUP_TEST_DIAGNOSTICS", previous)
+      else
+        System.delete_env("GROUP_TEST_DIAGNOSTICS")
+      end
+    end)
+
+    :ok
+  end
+
+  test "retains suite seeds and failure details", %{tmp_dir: directory} do
+    Group.TestDiagnostics.handle_cast({:suite_started, [seed: 12345]}, [])
+
+    failed = %ExUnit.Test{
+      name: :synthetic_failure,
+      module: __MODULE__,
+      state: {:failed, [{:error, %RuntimeError{message: "diagnostic fixture"}, []}]}
+    }
+
+    Group.TestDiagnostics.handle_cast({:test_finished, failed}, [])
+    [suite] = Path.wildcard(Path.join(directory, "*-suite.txt"))
+    [failure] = Path.wildcard(Path.join(directory, "*-failure.txt"))
+    assert File.read!(suite) =~ "12345"
+    assert File.read!(failure) =~ "diagnostic fixture"
+  end
+
+  test "snapshot includes topology, configuration, shard state and queue sizes" do
+    name = :"diagnostics_#{System.unique_integer([:positive])}"
+    start_supervised!({Group, name: name, shards: 1, log: false})
+    snapshot = Group.TestDiagnostics.snapshot()
+    assert snapshot.node == node()
+    assert snapshot.connected_nodes == Node.list()
+    group = Enum.find(snapshot.groups, &(&1.name == name))
+    assert group.config.num_shards == 1
+    [shard] = group.shards
+    assert shard.process[:message_queue_len] >= 0
+    assert shard.state =~ "pending_replicated_registry_len"
+    assert shard.state =~ "remote_shards"
+    assert snapshot.tables != []
+  end
+
+  test "unreachable peers produce diagnostics instead of hanging", %{tmp_dir: directory} do
+    peer = :"unreachable_diagnostic_#{System.unique_integer([:positive])}@127.0.0.1"
+    Group.TestDiagnostics.capture_peers([{nil, peer}])
+    [snapshot] = Path.wildcard(Path.join(directory, "*-peer_snapshot.txt"))
+    assert File.read!(snapshot) =~ "unavailable"
+  end
+end
diff --git a/test/distributed_test.exs b/test/distributed_test.exs
index 8694189..724a511 100644
--- a/test/distributed_test.exs
+++ b/test/distributed_test.exs
@@ -1,11 +1,17 @@
 defmodule Group.DistributedTest do
   use ExUnit.Case
 
+  @moduletag :distributed
   @moduletag :capture_log
   @moduletag timeout: 30_000
 
   alias Group.TestCluster
 
+  setup context do
+    Process.put(:group_test_context, Map.take(context, [:module, :test, :file, :line]))
+    :ok
+  end
+
   defp start_group_on_peers(peers, opts) do
     for {_pid, node} <- peers do
       TestCluster.start_group(node, opts)
@@ -590,14 +596,11 @@ defmodule Group.DistributedTest do
 
       start_group_on_peers(peers, opts)
 
-      # Wait for Erlang-level connectivity so disconnect_nodes actually works
-      TestCluster.assert_eventually(
-        fn ->
-          c_nodes = TestCluster.rpc!(node_c, Node, :list, [])
-          node_a in c_nodes and node_b in c_nodes
-        end,
-        timeout: 5000
-      )
+      # Finish discovery on every shard before cutting links. Erlang connectivity
+      # alone can leave handshake sends in flight that trigger reconnect retries.
+      TestCluster.assert_group_nodes(node_a, name, [node_b, node_c])
+      TestCluster.assert_group_nodes(node_b, name, [node_a, node_c])
+      TestCluster.assert_group_nodes(node_c, name, [node_a, node_b])
 
       # Set up nodedown monitor on A
       TestCluster.monitor_nodes_on(node_a, self())
@@ -609,8 +612,12 @@ defmodule Group.DistributedTest do
       # Wait for A to confirm it saw C go down
       assert_receive {:nodedown_on_remote, ^node_c}, 5000
 
+      # A separate node monitor does not establish that every shard handled DOWN.
+      TestCluster.assert_group_nodes(node_a, name, [node_b])
+      TestCluster.assert_group_nodes(node_b, name, [node_a])
+      TestCluster.assert_group_nodes(node_c, name, [])
+
       # While partitioned: register keys on A, join groups on C
-      # flush_shards ensures nodedown is processed before registering
       TestCluster.spawn_register(node_a, name, "user/from_a", %{origin: :a}, flush_shards: 2)
       TestCluster.spawn_join(node_c, name, "room/from_c", %{origin: :c})
 
@@ -2050,14 +2057,10 @@ defmodule Group.DistributedTest do
         length(nodes) >= 1
       end)
 
-      # Wait for Erlang-level connectivity so disconnect_nodes actually works
-      TestCluster.assert_eventually(
-        fn ->
-          c_nodes = TestCluster.rpc!(node_c, Node, :list, [])
-          node_a in c_nodes and node_b in c_nodes
-        end,
-        timeout: 5000
-      )
+      # Wait for Group discovery, not just Erlang connectivity, before partitioning.
+      TestCluster.assert_group_nodes(node_a, name, [node_b, node_c])
+      TestCluster.assert_group_nodes(node_b, name, [node_a, node_c])
+      TestCluster.assert_group_nodes(node_c, name, [node_a, node_b])
 
       # Set up nodedown monitors on A before partitioning
       TestCluster.monitor_nodes_on(node_a, self())
@@ -2070,9 +2073,12 @@ defmodule Group.DistributedTest do
       assert_receive {:nodedown_on_remote, ^node_c}, 5000
 
       # Wait for Group's own peer tables to reflect the partition before writing.
+      TestCluster.assert_group_nodes(node_a, name, [node_b])
+      TestCluster.assert_group_nodes(node_b, name, [node_a])
+      TestCluster.assert_group_nodes(node_c, name, [])
+
       TestCluster.assert_eventually(fn ->
-        node_c not in TestCluster.rpc!(node_a, Group, :nodes, [name]) and
-          node_c not in TestCluster.rpc!(node_a, Group, :nodes, [name, "game"])
+        node_c not in TestCluster.rpc!(node_a, Group, :nodes, [name, "game"])
       end)
 
       # Register data during partition
@@ -3304,9 +3310,24 @@ defmodule Group.DistributedTest do
 
       # Flap 3 times
       for _i <- 1..3 do
+        TestCluster.assert_group_nodes(node_a, name, [node_b])
+        TestCluster.assert_group_nodes(node_b, name, [node_a])
+        TestCluster.flush_shards(node_a, name)
+        TestCluster.flush_shards(node_b, name)
+
         TestCluster.disconnect_nodes(node_a, node_b)
         assert_receive {:nodedown_on_remote, ^node_b}, 5000
 
+        # A monitor notification can overtake shard cleanup. Do not let old
+        # replicated rows satisfy the re-sync assertion in the next cycle.
+        TestCluster.assert_group_nodes(node_a, name, [])
+        TestCluster.assert_group_nodes(node_b, name, [])
+
+        TestCluster.assert_eventually(fn ->
+          TestCluster.rpc!(node_b, Group, :lookup, [name, "stable/a"]) == nil and
+            TestCluster.rpc!(node_a, Group, :members, [name, "room/nil"]) == []
+        end)
+
         TestCluster.reconnect_nodes(node_a, node_b)
 
         # Wait for data to re-sync
diff --git a/test/group_test.exs b/test/group_test.exs
index b8476f5..1e7c277 100644
--- a/test/group_test.exs
+++ b/test/group_test.exs
@@ -2,16 +2,8 @@ defmodule GroupTest.ExtractMeta do
   def strip(meta), do: Map.take(meta, [:public])
 end
 
-defmodule GroupTest do
-  use ExUnit.Case, async: true
-
-  @moduletag :capture_log
-
-  setup do
-    name = :"test_group_#{System.unique_integer([:positive])}"
-    start_supervised!({Group, name: name, shards: 4, log: false})
-    {:ok, name: name}
-  end
+defmodule GroupTest.Startup do
+  use Group.LocalCase, async: true
 
   describe "startup options" do
     test "runtime config omits unused callback state", %{name: name} do
@@ -32,6 +24,11 @@ defmodule GroupTest do
       end
     end
   end
+end
+
+defmodule GroupTest.ProcessGroups do
+  # Ingress fairness assertions change VM-global trace patterns.
+  use Group.LocalCase, async: false
 
   describe "replica ingress fairness" do
     test "an oversized incoming batch yields to an already queued local write", %{name: name} do
@@ -227,6 +224,10 @@ defmodule GroupTest do
       [{_pid, %{v: 2}}] = Group.members(name, key)
     end
   end
+end
+
+defmodule GroupTest.Registry do
+  use Group.LocalCase, async: true
 
   describe "named-cluster mutation fencing" do
     test "a shard rejects registry and PG writes after their cluster epoch retires", %{name: name} do
@@ -376,6 +377,10 @@ defmodule GroupTest do
       assert Group.lookup(name, key) == nil
     end
   end
+end
+
+defmodule GroupTest.Membership do
+  use Group.LocalCase, async: true
 
   describe "members/2" do
     test "returns only joined processes", %{name: name} do
@@ -628,6 +633,10 @@ defmodule GroupTest do
       end
     end
   end
+end
+
+defmodule GroupTest.Monitoring do
+  use Group.LocalCase, async: true
 
   describe "self-events" do
     test "joining process receives its own :joined event if subscribed", %{name: name} do
@@ -756,6 +765,11 @@ defmodule GroupTest do
       refute_receive {:group, _, _}, 200
     end
   end
+end
+
+defmodule GroupTest.Clusters do
+  # Dispatch assertions change VM-global trace patterns.
+  use Group.LocalCase, async: false
 
   describe "named clusters" do
     test "connect and disconnect reject non-binary cluster names", %{name: name} do
@@ -1323,6 +1337,10 @@ defmodule GroupTest do
       refute_receive {:group, _, _}, 200
     end
   end
+end
+
+defmodule GroupTest.Requests do
+  use Group.LocalCase, async: true
 
   describe "call timeout option" do
     test "register honors timeout option" do
@@ -1539,6 +1557,10 @@ defmodule GroupTest do
       assert :ok = Group.TestCluster.assert_replica_consistent(name)
     end
   end
+end
+
+defmodule GroupTest.Fairness do
+  use Group.LocalCase, async: true
 
   describe "local request fairness" do
     test "a control flood yields to a queued local request after bounded work" do
@@ -1641,6 +1663,10 @@ defmodule GroupTest do
       assert Group.members(name, join_key2) == [{caller2, %{order: 2}}]
     end
   end
+end
+
+defmodule GroupTest.Consistency do
+  use Group.LocalCase, async: true
 
   describe "ETS table consistency" do
     test "tables are consistent after register + unregister", %{name: name} do
@@ -1781,6 +1807,11 @@ defmodule GroupTest do
       assert Group.TestCluster.assert_ets_consistent(name) == :ok
     end
   end
+end
+
+defmodule GroupTest.Counts do
+  # ETS call-count assertions change VM-global trace patterns.
+  use Group.LocalCase, async: false
 
   describe "local_registry_count/1" do
     test "local activity checks use bounded ETS selects", %{name: name} do
@@ -2046,6 +2077,10 @@ defmodule GroupTest do
         end)
     end
   end
+end
+
+defmodule GroupTest.Queries do
+  use Group.LocalCase, async: true
 
   describe "local_members/3" do
     test "returns local exact-key members and honors the limit", %{name: name} do
@@ -2207,6 +2242,10 @@ defmodule GroupTest do
                ])
     end
   end
+end
+
+defmodule GroupTest.Concurrency do
+  use Group.LocalCase, async: true
 
   describe "concurrent operations" do
     test "concurrent join/leave on same key doesn't produce duplicates", %{name: name} do
@@ -2273,6 +2312,11 @@ defmodule GroupTest do
       assert length(error_results) == 4
     end
   end
+end
+
+defmodule GroupTest.Events do
+  # Replication assertions change VM-global trace patterns.
+  use Group.LocalCase, async: false
 
   describe "event batching" do
     test "process death batches :unregistered and :left into one message", %{name: name} do
@@ -2413,6 +2457,10 @@ defmodule GroupTest do
       assert [%Group.Event{type: :unregistered}] = events
     end
   end
+end
+
+defmodule GroupTest.PGBuffering do
+  use Group.LocalCase, async: true
 
   describe "replicated PG receiver buffering" do
     test "legacy unsequenced ingress cannot materialize or delete rows" do
@@ -2452,6 +2500,10 @@ defmodule GroupTest do
       assert :ok = Group.TestCluster.assert_replica_consistent(name)
     end
   end
+end
+
+defmodule GroupTest.RegistryBuffering do
+  use Group.LocalCase, async: true
 
   describe "replica write-ahead journal" do
     test "concurrent shards retain independent append order", %{name: name} do
@@ -3159,164 +3211,4 @@ defmodule GroupTest do
       assert :ok = Group.TestCluster.assert_replica_consistent(name)
     end
   end
-
-  defp start_single_shard_group(opts \\ []) do
-    name = :"test_timeout_group_#{System.unique_integer([:positive])}"
-    opts = Keyword.merge([name: name, shards: 1, log: false], opts)
-    start_supervised!({Group, opts})
-    name
-  end
-
-  defp keys_for_shard(cluster, prefix, num_shards, shard, count) do
-    1
-    |> Stream.iterate(&(&1 + 1))
-    |> Stream.map(&"#{prefix}/#{&1}")
-    |> Stream.filter(&(Group.Replica.shard_index_for(cluster, &1, num_shards) == shard))
-    |> Enum.take(count)
-  end
-
-  defp suspend_only_shard(name) do
-    shard = Group.Replica.shard_name(name, 0)
-    :ok = :sys.suspend(shard)
-    shard
-  end
-
-  defp resume_shard_if_alive(shard) do
-    if Process.whereis(shard) do
-      :ok = :sys.resume(shard)
-    end
-
-    :ok
-  end
-
-  defp assert_genserver_call_timeout(fun) do
-    assert {:timeout, {GenServer, :call, _}} = catch_exit(fun.())
-  end
-
-  defp replicated_pg_join(cluster, key, pid, meta, reason) do
-    {:replicate_pg_batch,
-     [{:join, cluster, key, pid, meta, System.system_time(), reason, node(pid)}]}
-  end
-
-  defp replicated_register(cluster, key, pid, meta, _reason, time \\ System.system_time()) do
-    {:replicate_registry_batch, [{:register, cluster, key, pid, meta, time, node(pid)}]}
-  end
-
-  defp spawn_requester(fun, tag) do
-    parent = self()
-
-    spawn(fn ->
-      result = fun.()
-      send(parent, {tag, self(), result})
-      Process.sleep(:infinity)
-    end)
-  end
-
-  defp shard_message_queue_len(shard) do
-    case Process.info(Process.whereis(shard), :message_queue_len) do
-      {:message_queue_len, len} -> len
-      nil -> 0
-    end
-  end
-
-  defp flush_replicated_registry_barrier(shard) do
-    send(shard, {:group_dispatch, [self()], {:replicated_registry_buffer_flushed, shard}})
-  end
-
-  defp force_cluster_lease_sweep(name) do
-    lease_manager = Group.ClusterLease.lease_name(name)
-    send(lease_manager, :force_sweep)
-    :sys.get_state(lease_manager)
-    :ok
-  end
-
-  defp expire_cluster_lease(name, cluster) do
-    {ttl_ms, _expires_at} = Group.Replica.Data.cluster_lease(name, cluster)
-
-    Group.Replica.Data.put_cluster_lease(
-      name,
-      cluster,
-      ttl_ms,
-      System.monotonic_time(:millisecond) - 1
-    )
-
-    ttl_ms
-  end
-
-  defp spawn_forever do
-    spawn(fn -> Process.sleep(:infinity) end)
-  end
-
-  defp replica_ingress_fairness_owner(parent) do
-    receive do
-      {:write, shard, request} ->
-        ref = make_ref()
-        send(shard, {:group_local_request, self(), ref, request})
-        {reply, calls} = receive_local_write_with_trace(shard, ref, 0)
-        send(parent, {:local_write_finished, self(), reply, calls})
-        Process.sleep(:infinity)
-    end
-  end
-
-  defp membership_count_owner_loop do
-    receive do
-      {:membership_count_call, caller, ref, {:join, name, key, meta, opts}} ->
-        send(caller, {ref, Group.join(name, key, meta, opts)})
-        membership_count_owner_loop()
-
-      {:membership_count_call, caller, ref, {:leave, name, key, _meta, opts}} ->
-        send(caller, {ref, Group.leave(name, key, opts)})
-        membership_count_owner_loop()
-    end
-  end
-
-  defp membership_count_owner_call(owner, request) do
-    ref = make_ref()
-    send(owner, {:membership_count_call, self(), ref, request})
-
-    receive do
-      {^ref, result} -> result
-    after
-      1_000 -> flunk("membership count owner call timed out")
-    end
-  end
-
-  defp receive_local_write_with_trace(shard, ref, calls) do
-    receive do
-      {:trace, ^shard, :call,
-       {Group.Replica, :handle_replica_message, [_state, _source_node, _message]}} ->
-        receive_local_write_with_trace(shard, ref, calls + 1)
-
-      {:group_local_reply, ^ref, reply} ->
-        {reply, calls}
-    end
-  end
-
-  defp kill_if_alive(pid) do
-    if Process.alive?(pid) do
-      Process.exit(pid, :kill)
-    end
-
-    :ok
-  end
-
-  defp wait_until(fun, timeout \\ 1_000)
-
-  defp wait_until(fun, timeout) do
-    deadline = System.monotonic_time(:millisecond) + timeout
-    do_wait_until(fun, deadline)
-  end
-
-  defp do_wait_until(fun, deadline) do
-    if fun.() do
-      :ok
-    else
-      if System.monotonic_time(:millisecond) >= deadline do
-        flunk("condition did not become true")
-      end
-
-      Process.sleep(10)
-      do_wait_until(fun, deadline)
-    end
-  end
 end
diff --git a/test/history_test.exs b/test/history_test.exs
new file mode 100644
index 0000000..5dbc8f6
--- /dev/null
+++ b/test/history_test.exs
@@ -0,0 +1,87 @@
+defmodule Group.HistoryTest do
+  use ExUnit.Case, async: true
+
+  @moduletag :local
+  @moduletag :capture_log
+  @moduletag :history
+
+  # ExUnit seeds :rand per test. The complete generated input is retained in CI,
+  # so failures can be replayed with --seed, even when the campaign is larger.
+  for shards <- [1, 4, 8] do
+    @tag shards: shards
+    test "registry and PG histories match a reference model with #{shards} shards", context do
+      name = :"history_#{System.unique_integer([:positive])}"
+      start_supervised!({Group, name: name, shards: context.shards, log: false})
+      steps = String.to_integer(System.get_env("GROUP_HISTORY_STEPS", "100"))
+      assert steps in 1..100_000
+
+      history =
+        for step <- 1..steps do
+          {Enum.random([:register, :unregister, :join, :leave, :disconnect]),
+           Enum.random([nil, "red", "blue"]), "key/#{Enum.random(1..4)}", %{step: step}}
+        end
+
+      Group.TestDiagnostics.record(:history, %{
+        test: context.test,
+        shards: context.shards,
+        seed: ExUnit.configuration()[:seed],
+        operations: history
+      })
+
+      Enum.reduce(history, {%{}, %{}}, fn operation, model ->
+        model = apply_operation(name, operation, model)
+        assert_model(name, model, operation)
+        model
+      end)
+    end
+  end
+
+  defp apply_operation(_name, {:disconnect, nil, _key, _meta}, model), do: model
+
+  defp apply_operation(name, {:disconnect, cluster, _key, _meta}, {registry, groups}) do
+    :ok = Group.disconnect(name, cluster)
+    keep? = fn {{entry_cluster, _key}, _meta} -> entry_cluster != cluster end
+    {Map.filter(registry, keep?), Map.filter(groups, keep?)}
+  end
+
+  defp apply_operation(name, {operation, cluster, key, meta}, {registry, groups}) do
+    if cluster, do: Group.connect(name, cluster)
+    opts = [cluster: cluster]
+    entry = {cluster, key}
+
+    case operation do
+      :register ->
+        :ok = Group.register(name, key, meta, opts)
+        {Map.put(registry, entry, meta), groups}
+
+      :unregister ->
+        expected = if Map.has_key?(registry, entry), do: :ok, else: {:error, :undefined}
+        assert Group.unregister(name, key, opts) == expected
+        {Map.delete(registry, entry), groups}
+
+      :join ->
+        :ok = Group.join(name, key, meta, opts)
+        {registry, Map.put(groups, entry, meta)}
+
+      :leave ->
+        expected = if Map.has_key?(groups, entry), do: :ok, else: {:error, :not_in_group}
+        assert Group.leave(name, key, opts) == expected
+        {registry, Map.delete(groups, entry)}
+    end
+  end
+
+  defp assert_model(name, {registry, groups}, operation) do
+    for cluster <- [nil, "red", "blue"], key <- 1..4 do
+      key = "key/#{key}"
+      entry = {cluster, key}
+      expected_registration = if meta = registry[entry], do: {self(), meta}
+      expected_members = if meta = groups[entry], do: [{self(), meta}], else: []
+
+      assert Group.lookup(name, key, cluster: cluster) == expected_registration,
+             "registry mismatch after #{inspect(operation)}"
+
+      assert Group.members(name, key, cluster: cluster) == expected_members,
+             "PG mismatch after #{inspect(operation)}"
+    end
+  end
+end
diff --git a/test/jepsen/campaign.sh b/test/jepsen/campaign.sh
index 179d4cb..4e0a12c 100755
--- a/test/jepsen/campaign.sh
+++ b/test/jepsen/campaign.sh
@@ -13,6 +13,23 @@ recovery_time="${GROUP_JEPSEN_CAMPAIGN_RECOVERY:-15}"
 profile_grace="${GROUP_JEPSEN_CAMPAIGN_PROFILE_GRACE:-90}"
 artifact_dir="${GROUP_JEPSEN_CAMPAIGN_ARTIFACT_DIR:-}"
 
+transports=(distribution tcp chaos)
+scenarios=(mixed permanent)
+
+if [[ -n "${GROUP_JEPSEN_CAMPAIGN_TRANSPORT:-}" ]]; then
+  case "${GROUP_JEPSEN_CAMPAIGN_TRANSPORT}" in
+    distribution|tcp|chaos) transports=("${GROUP_JEPSEN_CAMPAIGN_TRANSPORT}") ;;
+    *) echo "invalid campaign transport" >&2; exit 1 ;;
+  esac
+fi
+
+if [[ -n "${GROUP_JEPSEN_CAMPAIGN_SCENARIO:-}" ]]; then
+  case "${GROUP_JEPSEN_CAMPAIGN_SCENARIO}" in
+    mixed|permanent) scenarios=("${GROUP_JEPSEN_CAMPAIGN_SCENARIO}") ;;
+    *) echo "invalid campaign scenario" >&2; exit 1 ;;
+  esac
+fi
+
 cd "${repo_dir}"
 
 if [[ "${GROUP_JEPSEN_SKIP_CHECKER:-0}" != "1" ]]; then
@@ -30,8 +47,8 @@ fi
 
 echo "Jepsen campaign artifacts: ${artifact_dir}"
 
-for transport in distribution tcp chaos; do
-  for scenario in mixed permanent; do
+for transport in "${transports[@]}"; do
+  for scenario in "${scenarios[@]}"; do
     log="${artifact_dir}/${transport}-${scenario}.log"
     sender_buffer_size=1
     min_delta_run_records=1
diff --git a/test/mutation/run.exs b/test/mutation/run.exs
index 4cec56e..96f1f14 100644
--- a/test/mutation/run.exs
+++ b/test/mutation/run.exs
@@ -19,7 +19,7 @@ defmodule Group.MutationCampaign do
         "        WireProtocol.stream_generation(stream_id) ==\n" <>
           "          Data.remote_generation(state.name, source_node) and",
       faulty_source: "        true and",
-      test: ["test/distributed_test.exs:5466"]
+      test: ["test/distributed_test.exs:5487"]
     },
     %{
       name: "accept_old_epoch",
@@ -28,7 +28,7 @@ defmodule Group.MutationCampaign do
         "        WireProtocol.stream_epoch(stream_id) ==\n" <>
           "          Data.remote_cluster_epoch(state.name, source_node, cluster) and",
       faulty_source: "        true and",
-      test: ["test/distributed_test.exs:4869"]
+      test: ["test/distributed_test.exs:4890"]
     },
     %{
       name: "advance_cursor_across_gap",
@@ -55,7 +55,7 @@ defmodule Group.MutationCampaign do
                   advertised_head
                 )
       """,
-      test: ["test/distributed_test.exs:5543"]
+      test: ["test/distributed_test.exs:5564"]
     },
     %{
       name: "registry_snapshot_is_additive",
@@ -70,7 +70,7 @@ defmodule Group.MutationCampaign do
       """,
       test: [
         "test/replica_snapshot_distributed_test.exs:16",
-        "test/distributed_test.exs:4073"
+        "test/distributed_test.exs:4094"
       ]
     },
     %{
@@ -80,7 +80,7 @@ defmodule Group.MutationCampaign do
       faulty_source: "          if Process.alive?(self()) do",
       test: [
         "test/replica_snapshot_distributed_test.exs:16",
-        "test/distributed_test.exs:4073"
+        "test/distributed_test.exs:4094"
       ]
     },
     %{
@@ -247,7 +247,7 @@ defmodule Group.MutationCampaign do
           "        append_process_down_records(state, reason_by_pid, pending_reg, pending_pg)\n",
       faulty_source:
         "      sequenced_downs =\n        if false,\n          do: append_process_down_records(state, reason_by_pid, pending_reg, pending_pg),\n          else: []\n",
-      test: ["test/distributed_test.exs:3983"]
+      test: ["test/distributed_test.exs:4004"]
     },
     %{
       name: "do_not_exit_conflict_loser",
@@ -310,7 +310,7 @@ defmodule Group.MutationCampaign do
         "         remote_replica_authority_hint(state.name, remote_node) == {generation, observed} do",
       faulty_source:
         "         elem(remote_replica_authority_hint(state.name, remote_node), 1) == observed do",
-      test: ["test/group_test.exs:3024"]
+      test: ["test/group_test.exs:3035"]
     },
     %{
       name: "install_incremental_after_newer_hint",
@@ -321,7 +321,7 @@ defmodule Group.MutationCampaign do
       faulty_source:
         "      Process.get(:ignore_incremental_authority_race, true) and\n" <>
           "      is_tuple(remote_replica_authority_hint(name, remote_node))\n",
-      test: ["test/group_test.exs:2958"]
+      test: ["test/group_test.exs:3035"]
     },
     %{
       name: "accept_hint_without_exact_authority",
@@ -357,7 +357,7 @@ defmodule Group.MutationCampaign do
           "      # crash in that window cannot strand the peer forever.\n" <>
           "      {{{:remote_authority_hint, :\"$1\"}, :_, :_}, [], [:\"$1\"]}\n",
       faulty_source: "      {{{:remote_view_info, shard, :\"$1\"}, :_, :_, :_}, [], [:\"$1\"]}\n",
-      test: ["test/group_test.exs:2897"]
+      test: ["test/group_test.exs:2974"]
     },
     %{
       name: "retain_retired_authority_repair",
@@ -385,7 +385,7 @@ defmodule Group.MutationCampaign do
       faulty_source: """
             :ok
       """,
-      test: ["test/distributed_test.exs:5688"]
+      test: ["test/distributed_test.exs:5709"]
     },
     %{
       name: "wait_for_periodic_lane_probe_after_authority_fanout",
@@ -474,7 +474,7 @@ defmodule Group.MutationCampaign do
         defp maybe_purge_remote_generation(state, _remote_node, _old_generation, _generation),
           do: state
       """,
-      test: ["test/distributed_test.exs:5688"]
+      test: ["test/distributed_test.exs:5709"]
     },
     %{
       name: "disable_periodic_heads",
@@ -488,21 +488,21 @@ defmodule Group.MutationCampaign do
           _ = state.peer_last_seen
           peers = []
       """,
-      test: ["test/distributed_test.exs:3983"]
+      test: ["test/distributed_test.exs:4059"]
     },
     %{
       name: "skip_journal_crash_repair",
       file: "lib/group/replica.ex",
       correct_source: ":ok = Data.repair_local_replica_journal(name, shard_index)",
       faulty_source: ":ok",
-      test: ["test/group_test.exs:2547"]
+      test: ["test/group_test.exs:2588"]
     },
     %{
       name: "skip_index_crash_repair",
       file: "lib/group/replica.ex",
       correct_source: ":ok = Data.repair_shard_indexes(name, shard_index)",
       faulty_source: ":ok",
-      test: ["test/group_test.exs:2593"]
+      test: ["test/group_test.exs:2670"]
     },
     %{
       name: "skip_pg_count_projection_update",
@@ -518,14 +518,14 @@ defmodule Group.MutationCampaign do
       faulty_source:
         "        _ = {table, count_key, total_delta, local_delta}\n" <>
           "        [total_count, local_count] = [0, 0]\n",
-      test: ["test/group_test.exs:1910"]
+      test: ["test/group_test.exs:1966"]
     },
     %{
       name: "retain_stale_pg_counts_on_shard_repair",
       file: "lib/group/replica/data.ex",
       correct_source: "    :ets.delete_all_objects(pg_counts)",
       faulty_source: "    _ = pg_counts",
-      test: ["test/group_test.exs:2593"]
+      test: ["test/group_test.exs:2670"]
     },
     %{
       name: "consult_stale_pg_counts_during_snapshot_repair",
@@ -565,7 +565,7 @@ defmodule Group.MutationCampaign do
           state = replay_local_journal(state)
           :ok = Data.repair_shard_indexes(name, shard_index)
       """,
-      test: ["test/group_test.exs:3094"]
+      test: ["test/group_test.exs:3171"]
     },
     %{
       name: "skip_inactive_cluster_repair",
@@ -575,7 +575,7 @@ defmodule Group.MutationCampaign do
         "    if Process.get(:run_primary_replica_repair, false),\n" <>
           "      do: repair_primary_replica_rows(name, shard),\n" <>
           "      else: :ok",
-      test: ["test/group_test.exs:2701"]
+      test: ["test/group_test.exs:2778"]
     },
     %{
       name: "skip_closed_cluster_completion",
@@ -591,7 +591,7 @@ defmodule Group.MutationCampaign do
       faulty_source: """
           _completed_clusters = []
       """,
-      test: ["test/group_test.exs:2701"]
+      test: ["test/group_test.exs:2778"]
     },
     %{
       name: "accept_unfenced_cluster_disconnect",
@@ -606,7 +606,7 @@ defmodule Group.MutationCampaign do
           "          Process.get(:accept_unfenced_cluster_disconnect, true)\n" <>
           "      end)\n\n" <>
           "    case epochs do\n",
-      test: ["test/group_test.exs:1485"]
+      test: ["test/group_test.exs:1528"]
     },
     %{
       name: "accept_completed_cluster_disconnect",
@@ -621,14 +621,14 @@ defmodule Group.MutationCampaign do
       faulty_source:
         "          _ = cluster\n" <>
           "          Process.get(:accept_completed_cluster_disconnect, true)",
-      test: ["test/group_test.exs:1485"]
+      test: ["test/group_test.exs:1528"]
     },
     %{
       name: "acknowledge_wrong_cluster_close_epoch",
       file: "lib/group/replica/data.ex",
       correct_source: "          [{^cluster, ^request_epoch, pending_shards}] ->\n",
       faulty_source: "          [{^cluster, _stored_epoch, pending_shards}] ->\n",
-      test: ["test/group_test.exs:2742"]
+      test: ["test/group_test.exs:2819"]
     },
     %{
       name: "accept_shared_authority_before_lane_install",
@@ -771,7 +771,7 @@ defmodule Group.MutationCampaign do
           "              :ets.delete(closed_local_cluster_epochs_table(state.name), cluster)\n",
       faulty_source:
         "              :ets.delete(closed_local_cluster_epochs_table(state.name), cluster)\n",
-      test: ["test/group_test.exs:2742"]
+      test: ["test/group_test.exs:2819"]
     },
     %{
       name: "retire_peer_authority_before_terminal_route_cleanup",
@@ -781,7 +781,7 @@ defmodule Group.MutationCampaign do
           "    :ok = delete_peer_routes(name, remote_node)\n",
       faulty_source:
         "    :ets.delete(replication_meta_table(name), {:remote_generation, remote_node})\n",
-      test: ["test/group_test.exs:2796"]
+      test: ["test/group_test.exs:2873"]
     },
     %{
       name: "stale_peer_cleanup_removes_rediscovered_routes",
@@ -793,7 +793,7 @@ defmodule Group.MutationCampaign do
         "    if Process.get(:purge_rediscovered_peer_routes, true) or\n" <>
           "         (is_nil(remote_generation(state.name, dead_node)) and\n" <>
           "            is_nil(remote_replica_authority_hint(state.name, dead_node))) do\n",
-      test: ["test/group_test.exs:2849"]
+      test: ["test/group_test.exs:2926"]
     },
     %{
       name: "stale_restart_cleanup_removes_reactivated_routes",
@@ -801,7 +801,7 @@ defmodule Group.MutationCampaign do
       correct_source:
         "      Enum.filter(clusters, &is_nil(local_cluster_epoch(state.name, &1)))\n",
       faulty_source: "      clusters\n",
-      test: ["test/group_test.exs:2776"]
+      test: ["test/group_test.exs:2853"]
     },
     %{
       name: "retain_authority_repair_after_nodedown",
@@ -812,7 +812,7 @@ defmodule Group.MutationCampaign do
       faulty_source:
         "        cluster_control_dirty: state.cluster_control_dirty,\n" <>
           "        authority_dirty_notified: MapSet.delete(state.authority_dirty_notified, dead_node)\n",
-      test: ["test/group_test.exs:2835"]
+      test: ["test/group_test.exs:2912"]
     },
     %{
       name: "retain_receive_cursor_for_inactive_local_cluster",
@@ -930,7 +930,7 @@ defmodule Group.MutationCampaign do
         {:noreply, state}
       end
       """,
-      test: ["test/group_test.exs:2393"]
+      test: ["test/group_test.exs:2466"]
     },
     %{
       name: "skip_cursorless_restart_authority_repair",
@@ -994,7 +994,7 @@ defmodule Group.MutationCampaign do
       file: "lib/group/replica.ex",
       correct_source: "    {turn, remaining} = Enum.split(messages, @incoming_batch_quota)",
       faulty_source: "    _ = @incoming_batch_quota\n    turn = messages\n    remaining = []",
-      test: ["test/group_test.exs:37"]
+      test: ["test/group_test.exs:34"]
     }
   ]
 
diff --git a/test/replica_model_property_test.exs b/test/replica_model_property_test.exs
index e4175d2..9fbcc7f 100644
--- a/test/replica_model_property_test.exs
+++ b/test/replica_model_property_test.exs
@@ -61,7 +61,7 @@ defmodule Group.ReplicaModelPropertyTest do
           end)
           |> ReplicaModelScheduler.stabilize_and_assert!()
 
-        assert scheduler.model != nil
+        assert is_map(scheduler.model.owners)
       after
         ReplicaModelScheduler.cleanup(scheduler)
       end
@@ -118,7 +118,7 @@ defmodule Group.ReplicaModelPropertyTest do
           end)
           |> ReplicaModelScheduler.stabilize_and_assert!()
 
-        assert scheduler.model != nil
+        assert is_map(scheduler.model.owners)
       after
         ReplicaModelScheduler.cleanup(scheduler)
       end
@@ -164,7 +164,7 @@ defmodule Group.ReplicaModelPropertyTest do
           |> run_schedule(after_restart)
           |> ReplicaModelScheduler.stabilize_and_assert!()
 
-        assert scheduler.model != nil
+        assert is_map(scheduler.model.owners)
       after
         ReplicaModelScheduler.cleanup(scheduler)
       end
@@ -221,7 +221,7 @@ defmodule Group.ReplicaModelPropertyTest do
           |> run_schedule(schedule)
           |> ReplicaModelScheduler.stabilize_and_assert!()
 
-        assert scheduler.model != nil
+        assert is_map(scheduler.model.owners)
       after
         ReplicaModelScheduler.cleanup(scheduler)
       end
@@ -281,7 +281,7 @@ defmodule Group.ReplicaModelPropertyTest do
           |> run_schedule(after_reopen)
           |> ReplicaModelScheduler.stabilize_and_assert!()
 
-        assert scheduler.model != nil
+        assert is_map(scheduler.model.owners)
       after
         ReplicaModelScheduler.cleanup(scheduler)
       end
diff --git a/test/support/local_case.ex b/test/support/local_case.ex
new file mode 100644
index 0000000..5760c86
--- /dev/null
+++ b/test/support/local_case.ex
@@ -0,0 +1,176 @@
+defmodule Group.LocalCase do
+  @moduledoc false
+  use ExUnit.CaseTemplate
+
+  using do
+    quote do
+      import Group.LocalCase
+      @moduletag :local
+      @moduletag :capture_log
+    end
+  end
+
+  setup do
+    name = :"test_group_#{System.unique_integer([:positive])}"
+    start_supervised!({Group, name: name, shards: 4, log: false})
+    {:ok, name: name}
+  end
+
+  def start_single_shard_group(opts \\ []) do
+    name = :"test_timeout_group_#{System.unique_integer([:positive])}"
+    opts = Keyword.merge([name: name, shards: 1, log: false], opts)
+    start_supervised!({Group, opts})
+    name
+  end
+
+  def keys_for_shard(cluster, prefix, num_shards, shard, count) do
+    1
+    |> Stream.iterate(&(&1 + 1))
+    |> Stream.map(&"#{prefix}/#{&1}")
+    |> Stream.filter(&(Group.Replica.shard_index_for(cluster, &1, num_shards) == shard))
+    |> Enum.take(count)
+  end
+
+  def replica_ingress_fairness_owner(parent) do
+    receive do
+      {:write, shard, request} ->
+        ref = make_ref()
+        send(shard, {:group_local_request, self(), ref, request})
+        {reply, calls} = receive_local_write_with_trace(shard, ref, 0)
+        send(parent, {:local_write_finished, self(), reply, calls})
+        Process.sleep(:infinity)
+    end
+  end
+
+  def membership_count_owner_loop do
+    receive do
+      {:membership_count_call, caller, ref, {:join, name, key, meta, opts}} ->
+        send(caller, {ref, Group.join(name, key, meta, opts)})
+        membership_count_owner_loop()
+
+      {:membership_count_call, caller, ref, {:leave, name, key, _meta, opts}} ->
+        send(caller, {ref, Group.leave(name, key, opts)})
+        membership_count_owner_loop()
+    end
+  end
+
+  def membership_count_owner_call(owner, request) do
+    ref = make_ref()
+    send(owner, {:membership_count_call, self(), ref, request})
+
+    receive do
+      {^ref, result} -> result
+    after
+      1_000 -> flunk("membership count owner call timed out")
+    end
+  end
+
+  defp receive_local_write_with_trace(shard, ref, calls) do
+    receive do
+      {:trace, ^shard, :call,
+       {Group.Replica, :handle_replica_message, [_state, _source_node, _message]}} ->
+        receive_local_write_with_trace(shard, ref, calls + 1)
+
+      {:group_local_reply, ^ref, reply} ->
+        {reply, calls}
+    end
+  end
+
+  def suspend_only_shard(name) do
+    shard = Group.Replica.shard_name(name, 0)
+    :ok = :sys.suspend(shard)
+    shard
+  end
+
+  def resume_shard_if_alive(shard) do
+    if Process.whereis(shard) do
+      :ok = :sys.resume(shard)
+    end
+
+    :ok
+  end
+
+  def assert_genserver_call_timeout(fun) do
+    assert {:timeout, {GenServer, :call, _}} = catch_exit(fun.())
+  end
+
+  def replicated_pg_join(cluster, key, pid, meta, reason) do
+    {:replicate_pg_batch,
+     [{:join, cluster, key, pid, meta, System.system_time(), reason, node(pid)}]}
+  end
+
+  def replicated_register(cluster, key, pid, meta, _reason, time \\ System.system_time()) do
+    {:replicate_registry_batch, [{:register, cluster, key, pid, meta, time, node(pid)}]}
+  end
+
+  def spawn_requester(fun, tag) do
+    parent = self()
+
+    spawn(fn ->
+      result = fun.()
+      send(parent, {tag, self(), result})
+      Process.sleep(:infinity)
+    end)
+  end
+
+  def shard_message_queue_len(shard) do
+    case Process.info(Process.whereis(shard), :message_queue_len) do
+      {:message_queue_len, len} -> len
+      nil -> 0
+    end
+  end
+
+  def flush_replicated_registry_barrier(shard) do
+    send(shard, {:group_dispatch, [self()], {:replicated_registry_buffer_flushed, shard}})
+  end
+
+  def force_cluster_lease_sweep(name) do
+    lease_manager = Group.ClusterLease.lease_name(name)
+    send(lease_manager, :force_sweep)
+    :sys.get_state(lease_manager)
+    :ok
+  end
+
+  def expire_cluster_lease(name, cluster) do
+    {ttl_ms, _expires_at} = Group.Replica.Data.cluster_lease(name, cluster)
+
+    Group.Replica.Data.put_cluster_lease(
+      name,
+      cluster,
+      ttl_ms,
+      System.monotonic_time(:millisecond) - 1
+    )
+
+    ttl_ms
+  end
+
+  def spawn_forever do
+    spawn(fn -> Process.sleep(:infinity) end)
+  end
+
+  def kill_if_alive(pid) do
+    if Process.alive?(pid) do
+      Process.exit(pid, :kill)
+    end
+
+    :ok
+  end
+
+  def wait_until(fun, timeout \\ 1_000) do
+    deadline = System.monotonic_time(:millisecond) + timeout
+    do_wait_until(fun, deadline)
+  end
+
+  defp do_wait_until(fun, deadline) do
+    if fun.() do
+      :ok
+    else
+      if System.monotonic_time(:millisecond) >= deadline do
+        flunk("condition did not become true")
+      end
+
+      Process.sleep(10)
+      do_wait_until(fun, deadline)
+    end
+  end
+end
diff --git a/test/support/test_cluster.ex b/test/support/test_cluster.ex
index 5ffa40a..697a16f 100644
--- a/test/support/test_cluster.ex
+++ b/test/support/test_cluster.ex
@@ -1,28 +1,49 @@
 defmodule Group.TestCluster do
   @moduledoc false
 
+  @doc "Start distribution only when a test actually needs real peers."
+  def ensure_distribution do
+    unless Node.alive?() do
+      epmd = System.find_executable("epmd") || raise "epmd executable not found"
+
+      case System.cmd(epmd, ["-daemon"], stderr_to_stdout: true) do
+        {_, 0} -> :ok
+        {output, status} -> raise "failed to start epmd (status #{status}): #{output}"
+      end
+
+      name = :"test_#{System.pid()}_#{System.unique_integer([:positive])}@127.0.0.1"
+      {:ok, _} = Node.start(name, :longnames)
+      Node.set_cookie(:group_test)
+    end
+
+    # Partition tests must keep their control connection to the test node.
+    :application.set_env(:kernel, :prevent_overlapping_partitions, false)
+    :ok
+  end
+
   @doc "Start N peer nodes with Group app loaded and ready"
   def start_peers(count, opts \\ []) do
+    ensure_distribution()
     cookie = Keyword.get(opts, :cookie, Node.get_cookie())
     code_paths = :code.get_path()
-    schedulers = Keyword.get(opts, :schedulers)
-
-    scheduler_args =
-      if schedulers, do: [~c"+S", ~c"#{schedulers}:#{schedulers}"], else: []
+    schedulers = Keyword.get(opts, :schedulers, System.get_env("GROUP_PEER_SCHEDULERS", "2"))
+    schedulers = String.to_integer(to_string(schedulers))
+    unless schedulers in 1..64, do: raise(ArgumentError, "peer schedulers must be in 1..64")
 
     args =
-      scheduler_args ++
-        [
-          ~c"-setcookie",
-          ~c"#{cookie}",
-          ~c"-kernel",
-          ~c"prevent_overlapping_partitions",
-          ~c"false"
-        ] ++
+      [
+        ~c"+S",
+        ~c"#{schedulers}:#{schedulers}",
+        ~c"-setcookie",
+        ~c"#{cookie}",
+        ~c"-kernel",
+        ~c"prevent_overlapping_partitions",
+        ~c"false"
+      ] ++
         Enum.flat_map(code_paths, fn p -> [~c"-pa", p] end)
 
     for _i <- 1..count do
-      name = :"peer#{System.unique_integer([:positive])}"
+      name = :"peer_#{System.pid()}_#{System.unique_integer([:positive])}"
 
       # A fixed inet_dist_listen_min/max inherited through ERL_AFLAGS makes
       # every child contend for the parent VM's distribution port. Peer args
@@ -38,11 +59,26 @@ defmodule Group.TestCluster do
 
       {:ok, _} = :rpc.call(node, :application, :ensure_all_started, [:elixir])
       {:ok, _} = :rpc.call(node, :application, :ensure_all_started, [:group])
+
+      Group.TestDiagnostics.record(:peer_started, %{
+        test: Process.get(:group_test_context),
+        node: node,
+        schedulers: schedulers
+      })
+
       {pid, node}
     end
   end
 
   def stop_peers(peers) do
+    try do
+      Group.TestDiagnostics.capture_peers(peers)
+    after
+      do_stop_peers(peers)
+    end
+  end
+
+  defp do_stop_peers(peers) do
     Enum.each(peers, fn {pid, _node} ->
       if pid do
         try do
@@ -496,10 +532,20 @@ defmodule Group.TestCluster do
   @doc "Monitor nodedown events from a remote node, forwarding to caller"
   def monitor_nodes_on(node, target_pid) do
     :erpc.call(node, fn ->
-      spawn(fn ->
-        :net_kernel.monitor_nodes(true)
-        forward_nodedown(target_pid)
-      end)
+      parent = self()
+
+      pid =
+        spawn(fn ->
+          :net_kernel.monitor_nodes(true)
+          send(parent, {:monitor_ready, self()})
+          forward_nodedown(target_pid)
+        end)
+
+      receive do
+        {:monitor_ready, ^pid} -> pid
+      after
+        5000 -> raise "monitor_nodes_on timed out"
+      end
     end)
   end
 
@@ -627,6 +673,23 @@ defmodule Group.TestCluster do
     :ok
   end
 
+  @doc "Wait for every shard and the shared peer table to agree on the expected Group peers."
+  def assert_group_nodes(node, name, expected_nodes) do
+    expected_nodes = Enum.sort(expected_nodes)
+
+    assert_eventually(fn ->
+      :erpc.call(node, fn ->
+        num_shards = Group.get_config(name).num_shards
+
+        Enum.sort(Group.nodes(name)) == expected_nodes and
+          Enum.all?(0..(num_shards - 1), fn shard ->
+            state = :sys.get_state(:"#{name}_replica_#{shard}")
+            Enum.sort(Map.keys(state.remote_shards)) == expected_nodes
+          end)
+      end)
+    end)
+  end
+
   @doc "Returns the current message_queue_len for a shard on a remote node."
   def shard_message_queue_len(node, name, shard) do
     :erpc.call(node, __MODULE__, :do_shard_message_queue_len, [name, shard])
diff --git a/test/support/test_diagnostics.ex b/test/support/test_diagnostics.ex
new file mode 100644
index 0000000..8ffe902
--- /dev/null
+++ b/test/support/test_diagnostics.ex
@@ -0,0 +1,98 @@
+defmodule Group.TestDiagnostics do
+  @moduledoc false
+  use GenServer
+
+  def init(opts), do: {:ok, opts}
+
+  def handle_cast({:suite_started, opts}, state) do
+    record(:suite, %{
+      options: opts,
+      elixir: System.version(),
+      otp: System.otp_release(),
+      schedulers: :erlang.system_info(:schedulers_online),
+      peer_schedulers: System.get_env("GROUP_PEER_SCHEDULERS", "2"),
+      history_steps: System.get_env("GROUP_HISTORY_STEPS", "100")
+    })
+
+    {:noreply, state}
+  end
+
+  def handle_cast({:test_finished, %ExUnit.Test{state: {:failed, _}} = test}, state) do
+    record(:failure, test)
+    {:noreply, state}
+  end
+
+  def handle_cast(_event, state), do: {:noreply, state}
+
+  def record(kind, data) do
+    if directory = System.get_env("GROUP_TEST_DIAGNOSTICS") do
+      File.mkdir_p!(directory)
+      id = "#{System.pid()}-#{System.unique_integer([:positive, :monotonic])}"
+      path = Path.join(directory, "#{id}-#{kind}.txt")
+      File.write!(path, inspect(data, pretty: true, limit: :infinity, printable_limit: :infinity))
+    end
+
+    :ok
+  end
+
+  def capture_peers(peers) do
+    if System.get_env("GROUP_TEST_DIAGNOSTICS") do
+      for {_pid, peer} <- peers do
+        snapshot =
+          try do
+            :erpc.call(peer, __MODULE__, :snapshot, [], 2_000)
+          catch
+            kind, reason -> %{unavailable: {kind, reason}}
+          end
+
+        record(:peer_snapshot, %{node: peer, snapshot: snapshot})
+      end
+    end
+  end
+
+  # Called before stopping fresh peers, including during failure/timeout cleanup.
+  # Bounded sys calls preserve evidence without turning a stuck shard into a hang.
+  def snapshot do
+    groups =
+      for {{Group, name}, config} <- :persistent_term.get(),
+          is_map(config),
+          is_integer(config[:num_shards]) do
+        shards =
+          for index <- 0..(config.num_shards - 1) do
+            shard = Group.Replica.shard_name(name, index)
+            pid = Process.whereis(shard)
+
+            state =
+              try do
+                :sys.get_state(shard, 100)
+                |> inspect(pretty: true, limit: 100, printable_limit: 8_000)
+              catch
+                kind, reason -> {kind, reason}
+              end
+
+            %{
+              shard: shard,
+              state: state,
+              process: pid && Process.info(pid, [:status, :current_function, :message_queue_len])
+            }
+          end
+
+        %{name: name, config: config, shards: shards}
+      end
+
+    %{
+      node: node(),
+      connected_nodes: Node.list(),
+      schedulers: :erlang.system_info(:schedulers_online),
+      groups: groups,
+      tables:
+        Enum.map(:ets.all(), fn table ->
+          :ets.info(table)
+          |> case do
+            :undefined -> :deleted
+            info -> Keyword.take(info, [:name, :size, :memory, :owner])
+          end
+        end)
+    }
+  end
+end
diff --git a/test/test_helper.exs b/test/test_helper.exs
index 15894c5..ddab61d 100644
--- a/test/test_helper.exs
+++ b/test/test_helper.exs
@@ -1,25 +1,5 @@
-# Start epmd and distribution if they are not already running (needed for distributed tests)
-epmd = System.find_executable("epmd") || raise "epmd executable not found"
-epmd_running? = fn -> match?({_, 0}, System.cmd(epmd, ["-names"], stderr_to_stdout: true)) end
-
-unless epmd_running?.() do
-  case System.cmd(epmd, ["-daemon"], stderr_to_stdout: true) do
-    {_, 0} -> :ok
-    {output, status} -> raise "failed to start epmd (status #{status}): #{output}"
-  end
-
-  unless epmd_running?.() do
-    raise "epmd did not become available after starting it"
-  end
-end
+ExUnit.start()
 
-unless Node.alive?() do
-  {:ok, _} = Node.start(:"test_#{System.unique_integer([:positive])}@127.0.0.1", :longnames)
-  Node.set_cookie(:group_test)
+if System.get_env("GROUP_TEST_DIAGNOSTICS") do
+  ExUnit.configure(formatters: [ExUnit.CLIFormatter, Group.TestDiagnostics])
 end
-
-# Disable :global's partition prevention to allow peer-to-peer disconnects
-# in distributed tests without the test node also being disconnected.
-:application.set_env(:kernel, :prevent_overlapping_partitions, false)
-
-ExUnit.start()