diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/Amethyst.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/Amethyst.kt
index 2e9b8fb209..45c757c32b 100644
--- a/amethyst/src/main/java/com/vitorpamplona/amethyst/Amethyst.kt
+++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/Amethyst.kt
@@ -56,11 +56,36 @@ import java.io.File
  */
 class Amethyst : Application() {
     init {
-        Log.minLevel = if (BuildConfig.DEBUG) LogLevel.DEBUG else LogLevel.ERROR
+        Log.minLevel = DEFAULT_LOG_LEVEL
         Log.d("AmethystApp") { "Creating App $this" }
     }
 
     companion object {
+        /**
+         * Restores the full firehose in debug builds: per-socket relay lifecycle
+         * (`Connecting…`/`Disconnected`/`OnOpen`), per-second event throughput, per-stream Arti
+         * SOCKS errors and per-relay census detail.
+         *
+         * Off by default. A cold start on the outbox model dials ~400 relays, and those sources
+         * alone emit ~3.7k of the ~6.2k lines an 80s boot produces — which buries the boot
+         * narrative and the relay-protocol warnings (`auth-required`, `rate-limited`,
+         * `unsupported: too many filters`) that are the actionable ones. Flip this, or set
+         * [Log.minLevel] at runtime, when you need the per-socket detail back.
+         */
+        const val VERBOSE_LOGS = false
+
+        /**
+         * Debug defaults to INFO so a boot reads as a narrative plus warnings; release defaults to
+         * WARN rather than ERROR so relay-protocol refusals stay visible in the field — they are
+         * the highest-value-per-line diagnostics we emit and were previously dropped entirely.
+         */
+        val DEFAULT_LOG_LEVEL: LogLevel =
+            when {
+                !BuildConfig.DEBUG -> LogLevel.WARN
+                VERBOSE_LOGS -> LogLevel.DEBUG
+                else -> LogLevel.INFO
+            }
+
         lateinit var instance: AppModules
             private set
     }
@@ -83,10 +108,15 @@ class Amethyst : Application() {
         // is self-contained (WebView + IPC), so we skip all app init there and
         // leave `instance` unset; any accidental use fails fast.
         if (isNappletSandbox) {
-            Log.d("AmethystApp") { "Skipping AppModules init in sandbox process" }
+            // Milestone, not chatter: this is the one line that explains why `instance` is unset
+            // and `LocalCache` is empty in this process. Without it a napplet-process log looks
+            // like a broken app rather than a deliberately secret-free sandbox.
+            Log.i("AmethystApp") { "Napplet sandbox process starting — no account, no AppModules" }
             return
         }
 
+        Log.i("AmethystApp") { "Amethyst ${BuildConfig.VERSION_NAME} starting in main process (log level ${Log.minLevel})" }
+
         instance = AppModules(this)
 
         // Hydrate the device-local favorite-apps list (main process only; the sandbox never reads it).
diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/LocalPreferences.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/LocalPreferences.kt
index 7ec8e71e50..90c9d1ce8a 100644
--- a/amethyst/src/main/java/com/vitorpamplona/amethyst/LocalPreferences.kt
+++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/LocalPreferences.kt
@@ -73,6 +73,7 @@ import com.vitorpamplona.quartz.nip78AppData.AppSpecificDataEvent
 import com.vitorpamplona.quartz.nip85TrustedAssertions.list.TrustProviderListEvent
 import com.vitorpamplona.quartz.nipB1Bolt12Zaps.offer.Bolt12OfferListEvent
 import com.vitorpamplona.quartz.utils.Log
+import com.vitorpamplona.quartz.utils.TimeUtils
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.async
@@ -325,6 +326,9 @@ object LocalPreferences {
                     }
 
                 if (!newSystemOfAccounts.isNullOrEmpty()) {
+                    // How many accounts are in play is the first thing you need when reading any
+                    // boot log: nearly every per-account subsystem below multiplies by this number.
+                    Log.i("LocalPreferences") { "Found ${newSystemOfAccounts.size} saved account(s)" }
                     newSystemOfAccounts
                 } else {
                     val oldAccounts = getString(PrefKeys.SAVED_ACCOUNTS, null)?.split(COMMA) ?: listOf()
@@ -712,6 +716,7 @@ object LocalPreferences {
 
     private suspend fun innerLoadCurrentAccountFromEncryptedStorage(npub: String?): AccountSettings? {
         Log.d("LocalPreferences") { "Load account from file $npub" }
+        val startedAtMs = TimeUtils.nowMillis()
         val result =
             withContext(Dispatchers.IO) {
                 return@withContext with(encryptedPreferences(npub)) {
@@ -1017,7 +1022,11 @@ object LocalPreferences {
                     )
                 }
             }
-        Log.d("LocalPreferences") { "Loaded account from file $npub" }
+        // Milestone with its cost attached. Decrypting and parsing one account's settings is one of
+        // the most expensive things a cold start does (it resolves a fan of backup events), it runs
+        // once per account, and "which account was slow" is the first question when a boot drags.
+        // The six intermediate steps above stay at DEBUG.
+        Log.i("LocalPreferences") { "Loaded account $npub in ${TimeUtils.nowMillis() - startedAtMs}ms" }
         return result
     }
 
diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/model/AntiSpamFilter.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/model/AntiSpamFilter.kt
index dbecbb436f..752492a5c3 100644
--- a/amethyst/src/main/java/com/vitorpamplona/amethyst/model/AntiSpamFilter.kt
+++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/model/AntiSpamFilter.kt
@@ -89,7 +89,13 @@ class AntiSpamFilter {
                 val link2 = njumpLink(NAddress.create(event.kind, event.pubKey, event.dTag(), relay))
                 val link1 = existingAddress?.let { njumpLink(NAddress.create(it.kind, it.pubKeyHex, it.dTag, relay)) } ?: link2
 
-                Log.w("Duplicated/SPAM") { "${relay?.url} $link1 $link2" }
+                // Debug, not warn: a duplicate detection is this filter working, not a fault, and
+                // it is already reported where it can be acted on — relayStats.newSpam below and
+                // the flowSpam emission that drives the UI. On a normal boot this fires ~34 times
+                // with a pair of njump links each, which is the widest line in the log and says
+                // nothing the spam counters don't. Keep the links at DEBUG for when you need to
+                // open the two events and compare them.
+                Log.d("Duplicated/SPAM") { "${relay?.url} $link1 $link2" }
 
                 // Log down offenders
                 val spammer = logOffender(hash, event)
@@ -118,7 +124,13 @@ class AntiSpamFilter {
                 // LRU cache while the spammer record still matches this hash.
                 val link1 = existingEvent?.let { njumpLink(NEvent.create(it, null, null, relay)) } ?: link2
 
-                Log.w("Duplicated/SPAM") { "${relay?.url} $link1 $link2" }
+                // Debug, not warn: a duplicate detection is this filter working, not a fault, and
+                // it is already reported where it can be acted on — relayStats.newSpam below and
+                // the flowSpam emission that drives the UI. On a normal boot this fires ~34 times
+                // with a pair of njump links each, which is the widest line in the log and says
+                // nothing the spam counters don't. Keep the links at DEBUG for when you need to
+                // open the two events and compare them.
+                Log.d("Duplicated/SPAM") { "${relay?.url} $link1 $link2" }
 
                 // Log down offenders
                 val spammer = logOffender(hash, event)
diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/diagnostics/BootRelayDiagnostics.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/diagnostics/BootRelayDiagnostics.kt
index cb21713277..1966bac786 100644
--- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/diagnostics/BootRelayDiagnostics.kt
+++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/relayClient/diagnostics/BootRelayDiagnostics.kt
@@ -56,7 +56,10 @@ import kotlin.concurrent.thread
  */
 class BootRelayDiagnostics(
     val client: INostrClient,
-    val dumpAtSeconds: List<Long> = listOf(20, 45, 90),
+    // 5s first: the pool is assembled and dialling well before 20s, and the early datapoint is what
+    // distinguishes "slow to connect" from "connected fine, slow to serve". Affordable because the
+    // rollup is now 3 INFO lines rather than 5 — the ===== banners moved to DEBUG.
+    val dumpAtSeconds: List<Long> = listOf(5, 20, 45, 90),
 ) {
     companion object {
         const val TAG = "BootRelayDiag"
@@ -197,8 +200,13 @@ class BootRelayDiagnostics(
     fun detach() = client.removeConnectionListener(listener)
 
     /**
-     * One line per relay plus a rollup. Kept to a single Log.w per line so the whole census
+     * One line per relay plus a rollup. Kept to a single Log call per line so the whole census
      * survives logcat's per-tag rate limiting on a busy boot.
+     *
+     * The rollup goes out at INFO — it is the one boot line worth reading by default, and it
+     * carries the aggregate that per-socket failure logging used to spell out a few hundred
+     * times. The per-relay WASTE/SERVE tables are DEBUG: useful when you are chasing a specific
+     * relay, too long (up to 45 lines a census) to sit in the default log.
      */
     fun dump(atSeconds: Long) {
         val snapshot = records.toMap()
@@ -214,27 +222,27 @@ class BootRelayDiagnostics(
             r.closed.forEach { (k, v) -> closedTotals[k] = (closedTotals[k] ?: 0) + v.get() }
         }
 
-        Log.w(TAG, "===== boot census @${atSeconds}s =====")
-        Log.w(
+        Log.d(TAG, "===== boot census @${atSeconds}s =====")
+        Log.i(
             TAG,
-            "pool=${snapshot.size} opened=${opened.size} served_events=${served.size} never_opened=${neverOpened.size} " +
+            "census @${atSeconds}s pool=${snapshot.size} opened=${opened.size} served_events=${served.size} never_opened=${neverOpened.size} " +
                 "dials=${snapshot.values.sumOf { it.tentatives.get() }} " +
                 "events=${snapshot.values.sumOf { it.events.get() }} " +
                 "reqs=${snapshot.values.sumOf { it.reqsSent.get() }} " +
                 "auths=${snapshot.values.sumOf { it.authsSent.get() }}",
         )
-        Log.w(TAG, "failures_by_cause=" + causeTotals.entries.sortedByDescending { it.value }.joinToString { "${it.key}:${it.value}" })
-        Log.w(TAG, "closed_by_prefix=" + closedTotals.entries.sortedByDescending { it.value }.joinToString { "${it.key}:${it.value}" })
+        Log.i(TAG, "census @${atSeconds}s failures_by_cause=" + causeTotals.entries.sortedByDescending { it.value }.joinToString { "${it.key}:${it.value}" })
+        Log.i(TAG, "census @${atSeconds}s closed_by_prefix=" + closedTotals.entries.sortedByDescending { it.value }.joinToString { "${it.key}:${it.value}" })
 
         // Relays that cost us dials and gave nothing back, worst first: the wasted-effort list.
-        Log.w(TAG, "--- top wasted dials (no events received) ---")
+        Log.d(TAG, "--- top wasted dials (no events received) ---")
         snapshot
             .filter { it.value.events.get() == 0 }
             .entries
             .sortedByDescending { it.value.tentatives.get() }
             .take(25)
             .forEach { (url, r) ->
-                Log.w(
+                Log.d(
                     TAG,
                     "WASTE ${url.url} dials=${r.tentatives.get()} opens=${r.opens.get()} " +
                         "fail=[${r.failures.entries.joinToString { "${it.key}:${it.value.get()}" }}] " +
@@ -245,17 +253,17 @@ class BootRelayDiagnostics(
 
         // The relays actually carrying the boot, so a suppression change can be checked for
         // coverage loss rather than just CLOSED reduction.
-        Log.w(TAG, "--- top event providers ---")
+        Log.d(TAG, "--- top event providers ---")
         served.entries
             .sortedByDescending { it.value.events.get() }
             .take(20)
             .forEach { (url, r) ->
-                Log.w(
+                Log.d(
                     TAG,
                     "SERVE ${url.url} events=${r.events.get()} reqs=${r.reqsSent.get()} eose=${r.eoses.get()} " +
                         "openMs=${r.firstOpenAtMs.get()} eoseMs=${r.firstEoseAtMs.get()} dials=${r.tentatives.get()}",
                 )
             }
-        Log.w(TAG, "===== end census @${atSeconds}s =====")
+        Log.d(TAG, "===== end census @${atSeconds}s =====")
     }
 }
diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/DecryptAndIndexProcessor.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/DecryptAndIndexProcessor.kt
index b8e81813fd..42f9e17572 100644
--- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/DecryptAndIndexProcessor.kt
+++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/DecryptAndIndexProcessor.kt
@@ -608,7 +608,12 @@ class GroupEventHandler(
             return
         }
         if (!manager.isMember(groupId)) {
-            Log.w("MarmotDbg") {
+            // Debug, not warn: relays serve kind:445 for every group they carry, so traffic for
+            // groups we are not in is the expected steady state, not an anomaly — unlike the null
+            // manager and missing 'h' tag above, which stay warnings because they mean we cannot
+            // process a group we *should* be handling. Fires ~22 times a boot (the same handful of
+            // events, re-offered on each pass), which drowns the two real warnings next to it.
+            Log.d("MarmotDbg") {
                 "GroupEventHandler.add: not a member of group=${groupId.take(8)}… — dropping kind:445 ${event.id.take(8)}…"
             }
             return
diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/tor/TorService.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/tor/TorService.kt
index 80fe1eacb7..d270967de4 100644
--- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/tor/TorService.kt
+++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/tor/TorService.kt
@@ -98,6 +98,24 @@ class TorService(
     private val _status = MutableStateFlow<TorServiceStatus>(TorServiceStatus.Off)
     override val status: StateFlow<TorServiceStatus> = _status.asStateFlow()
 
+    /**
+     * Every status change goes through here so the transition is logged exactly once, at INFO, with
+     * the time since bootstrap started.
+     *
+     * Tor state is the single biggest determinant of whether a boot works — a stuck `Connecting`,
+     * an `Active` that took 40s, and a silent drop back to `Off` are three completely different
+     * bugs that used to look identical in the log, because the per-status detail lived only in
+     * Arti's own DEBUG chatter. Self-transitions are not logged: the reset paths reassign `Off`
+     * defensively and repeating it adds nothing.
+     */
+    private fun setStatus(next: TorServiceStatus) {
+        val prev = _status.value
+        _status.value = next
+        if (prev::class == next::class && prev.toString() == next.toString()) return
+        val since = if (bootstrapStartedAtMs > 0) " after ${System.currentTimeMillis() - bootstrapStartedAtMs}ms" else ""
+        Log.i("TorService") { "${prev::class.simpleName} -> ${next::class.simpleName}$since" }
+    }
+
     /**
      * Runtime detector for a rotten guard sample. Arti logs [ALL_GUARDS_DOWN_MARKER] every time it
      * fails to find a usable guard; [GUARDS_DOWN_THRESHOLD] of those inside [GUARDS_DOWN_WINDOW_MS]
@@ -218,11 +236,11 @@ class TorService(
         lifecycleMutex.withLock {
             if (proxyRunning.get()) {
                 if (_status.value is TorServiceStatus.Active) return@withLock
-                _status.value = TorServiceStatus.Connecting
+                setStatus(TorServiceStatus.Connecting)
                 return@withLock
             }
 
-            _status.value = TorServiceStatus.Connecting
+            setStatus(TorServiceStatus.Connecting)
 
             withContext(Dispatchers.IO) {
                 // Initialize TorClient once — this bootstraps the Tor network.
@@ -292,7 +310,7 @@ class TorService(
                     if (initResult != 0) {
                         Log.e("TorService") { "Failed to initialize Arti on retry: error $initResult" }
                         initialized.set(false)
-                        _status.value = TorServiceStatus.Off
+                        setStatus(TorServiceStatus.Off)
                         return@withContext
                     }
                 }
@@ -313,7 +331,7 @@ class TorService(
 
                 if (!started) {
                     Log.e("TorService") { "Failed to start SOCKS proxy after $MAX_PORT_RETRIES attempts" }
-                    _status.value = TorServiceStatus.Off
+                    setStatus(TorServiceStatus.Off)
                     return@withContext
                 }
 
@@ -332,7 +350,7 @@ class TorService(
                 // reset/stop can't clobber it.
                 val startedAt = bootstrapStartedAtMs
                 val elapsed = if (startedAt > 0) System.currentTimeMillis() - startedAt else -1
-                _status.value = TorServiceStatus.Active(socksPort)
+                setStatus(TorServiceStatus.Active(socksPort))
                 Log.d("TorService") { "Arti SOCKS proxy active on port $socksPort (bootstrap took ${elapsed}ms)" }
             }
         }
@@ -350,7 +368,7 @@ class TorService(
                 Log.d("TorService") { "SOCKS proxy stopped" }
             }
 
-            _status.value = TorServiceStatus.Off
+            setStatus(TorServiceStatus.Off)
         }
     }
 
@@ -365,7 +383,7 @@ class TorService(
         lifecycleMutex.withLock {
             resetLocked()
         }
-        _status.value = TorServiceStatus.Off
+        setStatus(TorServiceStatus.Off)
     }
 
     /**
@@ -381,7 +399,7 @@ class TorService(
                 clearAllArtiData()
             }
         }
-        _status.value = TorServiceStatus.Off
+        setStatus(TorServiceStatus.Off)
     }
 
     /**
diff --git a/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/relayClient/assemblers/FeedMetadataCoordinatorTest.kt b/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/relayClient/assemblers/FeedMetadataCoordinatorTest.kt
index c0f8a609d4..4dc8837e8d 100644
--- a/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/relayClient/assemblers/FeedMetadataCoordinatorTest.kt
+++ b/commons/src/jvmTest/kotlin/com/vitorpamplona/amethyst/commons/relayClient/assemblers/FeedMetadataCoordinatorTest.kt
@@ -31,6 +31,7 @@ import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.job
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import org.junit.After
@@ -38,6 +39,9 @@ import org.junit.Assert.assertEquals
 import org.junit.Assert.assertTrue
 import org.junit.Before
 import org.junit.Test
+import java.util.Collections
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.atomic.AtomicInteger
 
 /**
  * Regression tests for PR #3483 review findings on FeedMetadataCoordinator:
@@ -73,34 +77,78 @@ class FeedMetadataCoordinatorTest {
 
     private fun pubkey(seed: Int): HexKey = seed.toString(16).padStart(64, '0')
 
+    /**
+     * Suspends until every coroutine the coordinator launched into [scope] has finished.
+     *
+     * This is what makes the assertions deterministic. The coordinator does its work in
+     * `scope.launch { subscribe(); gate.awaitAll(timeout); unsubscribe(); promote-or-roll-back }`,
+     * so "has call 1 finished?" is a question about the job tree, not about the clock. The tests
+     * used to answer it with `delay(timeoutMs + margin)` and assert straight after — which holds
+     * only while the dispatcher is free to start that coroutine promptly. On a loaded runner
+     * (macOS CI, 1431 tests in the same module) the launch itself can be queued past the margin,
+     * the roll-back lands late, the next call short-circuits, and the assertion fails on a
+     * perfectly healthy coordinator. Waiting on the jobs removes the margin entirely.
+     */
+    private suspend fun awaitCoordinatorIdle(timeoutMs: Long = 30_000) {
+        val deadline = System.currentTimeMillis() + timeoutMs
+        while (scope.coroutineContext.job.children
+                .any { it.isActive }
+        ) {
+            check(System.currentTimeMillis() < deadline) { "coordinator work did not settle within ${timeoutMs}ms" }
+            delay(2)
+        }
+    }
+
+    /** Polls [predicate] to a deadline. For preconditions we cannot express as job completion. */
+    private suspend fun waitUntil(
+        message: String,
+        timeoutMs: Long = 30_000,
+        predicate: () -> Boolean,
+    ) {
+        val deadline = System.currentTimeMillis() + timeoutMs
+        while (!predicate()) {
+            check(System.currentTimeMillis() < deadline) { "timed out waiting for: $message" }
+            delay(2)
+        }
+    }
+
     /**
      * Fake client that captures subscribe/unsubscribe and lets the test
      * drive EOSE notifications on any dispatcher we choose.
+     *
+     * Every collection here is written from the coordinator's coroutines (`Dispatchers.Default`,
+     * and `Dispatchers.IO` for the concurrent-EOSE test) and read from the test thread, so plain
+     * `mutableMapOf`/`mutableListOf` were two more races: an unsynchronized `size` read can be
+     * stale, and `fireEose` iterating `subscriptions.values` while a coordinator coroutine calls
+     * `unsubscribe` can throw ConcurrentModificationException. Concurrency is the thing under
+     * test here, so the fake must not be the weak link.
      */
     private class ControllableClient(
         private val delegate: INostrClient = EmptyNostrClient(),
     ) : INostrClient by delegate {
-        val subscriptions = mutableMapOf<String, SubscriptionListener?>()
-        val subscribeCalls = mutableListOf<Map<NormalizedRelayUrl, List<Filter>>>()
-        var unsubscribeCallCount = 0
-            private set
+        val subscriptions = ConcurrentHashMap<String, SubscriptionListener>()
+        val subscribeCalls: MutableList<Map<NormalizedRelayUrl, List<Filter>>> =
+            Collections.synchronizedList(mutableListOf())
+        private val unsubscribes = AtomicInteger(0)
+        val unsubscribeCallCount: Int get() = unsubscribes.get()
 
         override fun subscribe(
             subId: String,
             filters: Map<NormalizedRelayUrl, List<Filter>>,
             listener: SubscriptionListener?,
         ) {
-            subscriptions[subId] = listener
+            listener?.let { subscriptions[subId] = it }
             subscribeCalls.add(filters)
         }
 
         override fun unsubscribe(subId: String) {
             subscriptions.remove(subId)
-            unsubscribeCallCount++
+            unsubscribes.incrementAndGet()
         }
 
         fun fireEose(relay: NormalizedRelayUrl) {
-            subscriptions.values.filterNotNull().forEach { it.onEose(relay, forFilters = null) }
+            // Snapshot: a coordinator coroutine may unsubscribe concurrently.
+            subscriptions.values.toList().forEach { it.onEose(relay, forFilters = null) }
         }
     }
 
@@ -119,13 +167,13 @@ class FeedMetadataCoordinatorTest {
 
             // Call 1 — no relay EOSEs; must time out.
             coordinator.loadKind3Batched(pubkeys, timeoutMs = 200)
-            delay(350) // exceed the timeout
+            awaitCoordinatorIdle() // call 1 timed out and rolled back
 
             // Call 2 — the same pubkeys must be re-subscribed since call 1
             // never got a successful EOSE. The old code would silently
             // short-circuit here.
             coordinator.loadKind3Batched(pubkeys, timeoutMs = 200)
-            delay(50) // let the launcher run
+            awaitCoordinatorIdle()
 
             assertEquals(
                 "Zero-EOSE timeout must not permanently dedup pubkeys",
@@ -158,13 +206,13 @@ class FeedMetadataCoordinatorTest {
             val pubkeys = listOf(pubkey(1), pubkey(2))
 
             coordinator.loadKind3Batched(pubkeys, timeoutMs = 1_000)
-            // Give the launcher time to register the listener before we fire.
-            delay(50)
+            // The listener must be registered before we fire, or the EOSEs go nowhere.
+            waitUntil("subscription registered") { client.subscriptions.isNotEmpty() }
             indexRelays.forEach(client::fireEose)
-            delay(200) // let the coordinator finish + promote to queued
+            awaitCoordinatorIdle() // coordinator finished + promoted to queued
 
             coordinator.loadKind3Batched(pubkeys, timeoutMs = 200)
-            delay(50)
+            awaitCoordinatorIdle()
 
             assertEquals(
                 "Successful call must dedup subsequent identical calls",
@@ -187,13 +235,13 @@ class FeedMetadataCoordinatorTest {
             val pubkeys = listOf(pubkey(1))
 
             coordinator.loadKind3Batched(pubkeys, timeoutMs = 300)
-            delay(30)
+            waitUntil("subscription registered") { client.subscriptions.isNotEmpty() }
             // Only 1 of 3 EOSEs — timeout still fires but we made progress.
             client.fireEose(relay1)
-            delay(400)
+            awaitCoordinatorIdle()
 
             coordinator.loadKind3Batched(pubkeys, timeoutMs = 200)
-            delay(50)
+            awaitCoordinatorIdle()
 
             assertEquals(
                 "≥1 EOSE = progress = promote to queued (avoid re-asking)",
@@ -222,7 +270,7 @@ class FeedMetadataCoordinatorTest {
                 )
 
             coordinator.loadKind3Batched(listOf(pubkey(1)), timeoutMs = 2_000)
-            delay(50) // wait for subscription
+            waitUntil("subscription registered") { client.subscriptions.isNotEmpty() }
 
             // Fire EOSEs concurrently from many dispatchers.
             val jobs =
@@ -235,9 +283,9 @@ class FeedMetadataCoordinatorTest {
 
             // The 2nd call must short-circuit — every relay EOSE'd, so
             // pubkey(1) is now in queuedKind3Pubkeys.
-            delay(100)
+            awaitCoordinatorIdle()
             coordinator.loadKind3Batched(listOf(pubkey(1)), timeoutMs = 200)
-            delay(50)
+            awaitCoordinatorIdle()
 
             assertEquals(
                 "Under concurrent EOSE from all relays, aggregator must reach target",
@@ -261,10 +309,10 @@ class FeedMetadataCoordinatorTest {
 
             // Call 1 — zero EOSE, timeout.
             coordinator.loadMetadataBatched(pubkeys, timeoutMs = 200)
-            delay(350)
+            awaitCoordinatorIdle()
             // Call 2 — must re-subscribe.
             coordinator.loadMetadataBatched(pubkeys, timeoutMs = 200)
-            delay(50)
+            awaitCoordinatorIdle()
 
             assertTrue(
                 "Metadata batch also retries on zero-EOSE timeout",
@@ -284,13 +332,13 @@ class FeedMetadataCoordinatorTest {
                 )
 
             coordinator.loadKind3Batched(listOf(pubkey(1)), timeoutMs = 200)
-            delay(50)
+            waitUntil("subscription registered") { client.subscriptions.isNotEmpty() }
             // clear() must drop the in-flight tracker even mid-request.
             coordinator.clear()
-            delay(300) // let call 1 finish + roll back
+            awaitCoordinatorIdle() // call 1 finished + rolled back
 
             coordinator.loadKind3Batched(listOf(pubkey(1)), timeoutMs = 200)
-            delay(50)
+            awaitCoordinatorIdle()
 
             assertTrue(client.subscribeCalls.size >= 2)
         }
diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/concord/cord02Community/ConcordCommunityState.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/concord/cord02Community/ConcordCommunityState.kt
index 31c74caa7e..b2c8db0179 100644
--- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/concord/cord02Community/ConcordCommunityState.kt
+++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/concord/cord02Community/ConcordCommunityState.kt
@@ -89,7 +89,13 @@ data class ConcordCommunityState(
             ownerPubKey: String,
             floors: Map<String, EntityFloor> = emptyMap(),
         ): Map<String, EntityFloor> {
-            val pool = EditionFold.admissible(editions, floors)
+            // This call folds exactly one epoch's editions, so they ARE the snapshot: an entity
+            // appearing here has been re-wrapped into this epoch and must anchor on version, not
+            // on a `prev` that necessarily dangles back into the epoch the floor came from. The
+            // known heads `admissible` re-seats are not in the set, so an entity this epoch never
+            // mentioned correctly keeps chain-walk semantics.
+            val snapshot = editions.mapTo(HashSet(editions.size)) { it.rumorId }
+            val pool = EditionFold.admissible(editions, floors, snapshot = snapshot)
             val authority = AuthorityResolver.resolve(pool, ownerPubKey)
             val out = HashMap<String, EntityFloor>(floors)
             for ((kind, list) in pool.groupBy { it.entityKind }) {
@@ -97,7 +103,7 @@ data class ConcordCommunityState(
                 // Gate the CANDIDATES, don't pre-filter the chain: a rejected edition mid-chain must
                 // stay inert instead of orphaning the authorized editions above it (EditionFold.candidates).
                 val heads =
-                    EditionFold.foldGated(list, floors) {
+                    EditionFold.foldGated(list, floors, snapshot = snapshot) {
                         authority.isOwner(it.author) || (bit != null && authority.hasPermission(it.author, bit))
                     }
                 for ((entity, head) in heads) {
@@ -119,10 +125,15 @@ data class ConcordCommunityState(
             // authority chains, the per-kind gated folds), so the anti-rollback floor is applied
             // once, up front, on the shared pool: a rolled-back edition is never seen by any of
             // them, and the head we already folded is re-seated so the entity keeps its state.
-            @Suppress("NAME_SHADOWING")
-            val editions = EditionFold.admissible(editions, floors)
+            // The editions handed in are one epoch's — the current one — so they are the
+            // snapshot that selects the compaction arm. Captured BEFORE `admissible` re-seats
+            // known heads from older epochs, which must not be mistaken for this epoch's.
+            val snapshot = editions.mapTo(HashSet(editions.size)) { it.rumorId }
 
-            val heads = EditionFold.fold(editions, floors).values
+            @Suppress("NAME_SHADOWING")
+            val editions = EditionFold.admissible(editions, floors, snapshot = snapshot)
+
+            val heads = EditionFold.fold(editions, floors, snapshot = snapshot).values
             // Resolve authority from the FULL edition set (not the structural heads): the resolver
             // folds each role/grant chain through authorized editions only, so a rogue higher-version
             // edition can't supersede a legit one before authority is even judged.
@@ -142,7 +153,7 @@ data class ConcordCommunityState(
                 kind: ControlEntityKind,
                 bit: Int,
             ): Map<String, ControlEdition> =
-                EditionFold.foldGated(editions.filter { it.entityKind == kind }, floors) {
+                EditionFold.foldGated(editions.filter { it.entityKind == kind }, floors, snapshot = snapshot) {
                     authority.isOwner(it.author) || authority.hasPermission(it.author, bit)
                 }
 
diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/concord/cord04Roles/EditionFold.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/concord/cord04Roles/EditionFold.kt
index 5d37555d66..4e5b8fa72c 100644
--- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/concord/cord04Roles/EditionFold.kt
+++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/concord/cord04Roles/EditionFold.kt
@@ -22,6 +22,7 @@ package com.vitorpamplona.quartz.concord.cord04Roles
 
 import com.vitorpamplona.quartz.nip01Core.core.toHexKey
 import com.vitorpamplona.quartz.utils.Log
+import com.vitorpamplona.quartz.utils.concurrent.ConcurrentSet
 
 /**
  * The anti-rollback floor for one Control Plane entity: the [version] and
@@ -97,16 +98,61 @@ typealias GapReporter = (entityIdHex: String, floorVersion: Long, offeredVersion
 object EditionFold {
     private const val TAG = "ConcordEditionFold"
 
+    /**
+     * Rollback refusals already reported by [LOG_GAP], keyed by the exact refusal
+     * (entity + floor + offered version) rather than by entity, so a *new* rollback
+     * attempt against the same entity is still reported.
+     *
+     * Bounded in practice: a key is only ever added when an entity is offered a chain
+     * it already refused at a version pair it has not seen, which happens once per real
+     * rotation, not once per fold. Deliberately never cleared and deliberately
+     * fail-open — see [LOG_GAP].
+     */
+    private val reportedGaps = ConcurrentSet<String>()
+
+    /** Above this many distinct refusals, stop deduplicating and just warn. See [LOG_GAP]. */
+    private const val MAX_TRACKED_GAPS = 4096
+
     /**
      * The default [GapReporter]: a rollback refusal is security-relevant (a rotator
      * tried to revert an entity), so it is warned, never swallowed.
+     *
+     * Warned **once per distinct refusal**, though. Amethyst re-folds the whole buffer
+     * from scratch on every control-plane change, so an entity stuck refusing a rollback
+     * re-reports the identical refusal on every refold — 22 byte-identical warnings in an
+     * 80s cold start, which reads as 22 attacks rather than one unchanged state. The
+     * dedup is on the message's own contents, so nothing a reader could act on is lost.
+     *
+     * Fails **open**: past [MAX_TRACKED_GAPS] distinct refusals it reverts to warning
+     * every time. A flood of distinct rollback attempts is exactly when the warnings
+     * matter most, so the failure mode is a noisy log, never a silenced one.
      */
     val LOG_GAP: GapReporter = { entityIdHex, floorVersion, offeredVersion ->
-        Log.w(TAG) {
-            "Control-plane rollback refused for entity $entityIdHex: already folded v$floorVersion, offered chain tops out at v$offeredVersion and does not connect to it"
+        val firstReport =
+            reportedGaps.size() >= MAX_TRACKED_GAPS ||
+                reportedGaps.add("$entityIdHex@$floorVersion<-$offeredVersion")
+
+        if (firstReport) {
+            Log.w(TAG) {
+                "Control-plane rollback refused for entity $entityIdHex: already folded v$floorVersion, offered chain tops out at v$offeredVersion and does not connect to it"
+            }
         }
     }
 
+    /**
+     * The compaction-era head: the highest-version edition at or above [floorVersion], ties
+     * broken by the lower rumor id — no `prev`, no hash, no contiguity. See the compaction arm
+     * in [foldEntity] for why version is the right (and only sound) anchor behind a Refounding.
+     * Null when nothing at or above the floor was offered, which is the only gap this arm has.
+     */
+    private fun bootstrapHead(
+        editions: List<ControlEdition>,
+        floorVersion: Long,
+    ): ControlEdition? =
+        editions
+            .filter { it.version >= floorVersion }
+            .minWithOrNull(compareByDescending<ControlEdition> { it.version }.thenBy { it.rumorId })
+
     /**
      * Groups mixed [editions] by entity id and folds each to its head, honoring the
      * per-entity anti-rollback [floors] (keyed by [ControlEdition.entityIdHex]).
@@ -119,12 +165,13 @@ object EditionFold {
     fun fold(
         editions: Collection<ControlEdition>,
         floors: Map<String, EntityFloor> = emptyMap(),
+        snapshot: Set<String>? = null,
         onGap: GapReporter = LOG_GAP,
     ): Map<String, ControlEdition> {
         val byEntity = editions.groupBy { it.entityIdHex }
         val out = HashMap<String, ControlEdition>(byEntity.size)
         for ((entity, list) in byEntity) {
-            foldEntity(list, floors[entity], onGap)?.let { out[entity] = it }
+            foldEntity(list, floors[entity], snapshot, onGap)?.let { out[entity] = it }
         }
         return out
     }
@@ -134,18 +181,44 @@ object EditionFold {
      *
      * With no [floor] this is the fresh-joiner fold: genesis-anchored, falling back
      * to the lowest-version edition present (the compaction bootstrap). With a
-     * [floor] the walk is anchored at the exact edition already folded; if that
-     * edition is not among [editions] the chain is **gapped** and nothing above the
-     * floor is adopted — [EntityFloor.known] is kept instead (or null when we no
-     * longer hold it). A head is therefore never below the floor version.
+     * [floor] the walk is anchored at the edition already folded **or** at its
+     * immediate successor when that successor cites the floor's hash; failing both
+     * the chain is **gapped** and nothing above the floor is adopted —
+     * [EntityFloor.known] is kept instead (or null when we no longer hold it). A head
+     * is therefore never below the floor version.
+     *
+     * [snapshot] is the set of [ControlEdition.rumorId]s belonging to the epoch being
+     * folded, supplied once a community has Refounded. An entity present in it takes
+     * the version-anchored compaction arm instead of the chain walk — see the arm
+     * itself for why. Null (the default) keeps the pure chain walk, which is right for
+     * a single-epoch fold and for every caller that has no epoch to speak of.
      */
     fun foldEntity(
         editions: List<ControlEdition>,
         floor: EntityFloor? = null,
+        snapshot: Set<String>? = null,
         onGap: GapReporter = LOG_GAP,
     ): ControlEdition? {
         if (editions.isEmpty()) return floor?.known
 
+        // Compaction arm (CORD-06 §3). Once this entity has been re-wrapped into the epoch
+        // being folded, `prev` chaining across the epoch boundary is meaningless: a Refounding
+        // trims history and re-wraps the head alone, so the predecessor our floor names stays
+        // behind on the older epoch and every offered `prev` dangles by design. Anchor on
+        // VERSION instead — a re-wrap preserves the original author's signature but cannot
+        // raise the version inside the signed seal, so a re-served stale edition always loses
+        // to the compacted head. Presence of the entity in the snapshot selects the ARM;
+        // version selects the HEAD, over every edition we hold and not just the subset.
+        if (floor != null && snapshot != null && editions.any { it.rumorId in snapshot }) {
+            return bootstrapHead(editions, floor.version)
+                ?: run {
+                    // Nothing at or above the floor was served: the head we already accepted
+                    // vanished from the offered set — withheld, so fail closed.
+                    onGap(editions[0].entityIdHex, floor.version, editions.maxOf { it.version })
+                    floor.known
+                }
+        }
+
         // Index editions by version, keeping the tie-break winner where several
         // share a version (lower rumor id wins).
         val byVersion = HashMap<Long, MutableList<ControlEdition>>()
@@ -153,12 +226,25 @@ object EditionFold {
 
         var head =
             if (floor != null) {
-                // Anchored at what we already folded: the offered set MUST contain that exact
-                // edition (same version AND same hash — a same-version sibling is a fork, not
-                // our chain). Failing that, refuse to move at all rather than accept an
-                // unverifiable jump; walking up from the floor also makes a head below the
-                // floor version structurally impossible.
-                editions.firstOrNull { it.version == floor.version && it.hashHex == floor.hashHex }
+                // Anchored at what we already folded. Below the floor is history we absorbed, so
+                // the anchor is the lowest offered version at or above it, and only two shapes
+                // connect: that edition IS the floor (same version AND hash — a same-version
+                // sibling is a fork, not our chain), or it is the floor's immediate successor and
+                // cites the floor's hash. The latter is the ordinary cross-epoch shape and is a
+                // STRONGER proof of connection than mere presence. Anything else is a jump we
+                // refuse; walking up from the anchor also makes a head below the floor version
+                // structurally impossible.
+                val lowest = byVersion.keys.filter { it >= floor.version }.minOrNull()
+                val winner = lowest?.let { v -> byVersion[v]?.minByOrNull { it.rumorId } }
+                val anchor =
+                    when {
+                        winner == null -> null
+                        lowest == floor.version -> winner.takeIf { it.hashHex == floor.hashHex }
+                        lowest == floor.version + 1 ->
+                            winner.takeIf { it.prevHash != null && it.prevHash.toHexKey() == floor.hashHex }
+                        else -> null
+                    }
+                anchor
                     ?: run {
                         onGap(editions[0].entityIdHex, floor.version, editions.maxOf { it.version })
                         return floor.known
@@ -222,12 +308,21 @@ object EditionFold {
     fun candidates(
         editions: List<ControlEdition>,
         floor: EntityFloor? = null,
+        snapshot: Set<String>? = null,
         onGap: GapReporter = LOG_GAP,
     ): List<ControlEdition> {
-        val head = foldEntity(editions, floor, onGap) ?: return emptyList()
+        // Ask the fold whether it gapped rather than re-deriving the condition here: with the
+        // compaction arm and the successor anchor there are three ways to connect, and a second
+        // copy of that test is a bug waiting to drift out of sync with the first.
+        var gapped = false
+        val head =
+            foldEntity(editions, floor, snapshot) { e, f, o ->
+                gapped = true
+                onGap(e, f, o)
+            } ?: return emptyList()
         // A gap re-seated the known head: nothing from the offered set is admissible above
         // the floor, so the known edition is the only candidate.
-        if (floor != null && editions.none { it.version == floor.version && it.hashHex == floor.hashHex }) {
+        if (floor != null && gapped) {
             return listOf(head)
         }
         val out = ArrayList<ControlEdition>(editions.size)
@@ -247,9 +342,10 @@ object EditionFold {
     fun foldEntityGated(
         editions: List<ControlEdition>,
         floor: EntityFloor? = null,
+        snapshot: Set<String>? = null,
         onGap: GapReporter = LOG_GAP,
         gate: (ControlEdition) -> Boolean,
-    ): ControlEdition? = candidates(editions, floor, onGap).firstOrNull(gate)
+    ): ControlEdition? = candidates(editions, floor, snapshot, onGap).firstOrNull(gate)
 
     /**
      * Groups mixed [editions] by entity id and folds each to the highest-priority
@@ -258,13 +354,14 @@ object EditionFold {
     fun foldGated(
         editions: Collection<ControlEdition>,
         floors: Map<String, EntityFloor> = emptyMap(),
+        snapshot: Set<String>? = null,
         onGap: GapReporter = LOG_GAP,
         gate: (ControlEdition) -> Boolean,
     ): Map<String, ControlEdition> {
         val byEntity = editions.groupBy { it.entityIdHex }
         val out = HashMap<String, ControlEdition>(byEntity.size)
         for ((entity, list) in byEntity) {
-            foldEntityGated(list, floors[entity], onGap, gate)?.let { out[entity] = it }
+            foldEntityGated(list, floors[entity], snapshot, onGap, gate)?.let { out[entity] = it }
         }
         return out
     }
@@ -285,6 +382,7 @@ object EditionFold {
     fun admissible(
         editions: Collection<ControlEdition>,
         floors: Map<String, EntityFloor>,
+        snapshot: Set<String>? = null,
         onGap: GapReporter = LOG_GAP,
     ): List<ControlEdition> {
         if (floors.isEmpty()) return editions.toList()
@@ -298,7 +396,12 @@ object EditionFold {
                 out.addAll(list)
                 continue
             }
-            if (list.any { it.version == floor.version && it.hashHex == floor.hashHex }) {
+            // Same three-way connection test as the fold, asked of the fold itself so the two
+            // can't drift apart: present at the floor, a successor citing it, or the compaction
+            // arm's version anchor.
+            var gapped = false
+            foldEntity(list, floor, snapshot) { _, _, _ -> gapped = true }
+            if (!gapped) {
                 out.addAll(list)
                 continue
             }
diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/RelayLogger.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/RelayLogger.kt
index f1b95de2f2..78b304edae 100644
--- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/RelayLogger.kt
+++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/accessories/RelayLogger.kt
@@ -101,12 +101,20 @@ class RelayLogger(
                 Log.d(logTag(relay.url), "Disconnected")
             }
 
+            /**
+             * Debug, not error: under the outbox model a client dials hundreds of relays per boot,
+             * and a large share of them are dead, unreachable, or refused by the local Tor proxy.
+             * One line per failed socket is a few hundred lines that say the same four things
+             * ("SOCKS: TTL expired", "SSLHandshakeException", "Host unreachable", …) and cannot be
+             * acted on individually. The actionable form is the aggregate — failures bucketed by
+             * cause, which the consumer's boot census already reports.
+             */
             override fun onCannotConnect(
                 relay: IRelayClient,
                 errorMessage: String,
             ) {
                 super.onCannotConnect(relay, errorMessage)
-                Log.e(logTag(relay.url), errorMessage)
+                Log.d(logTag(relay.url), errorMessage)
             }
         }
 
diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/concord/cord04Roles/EditionFoldFloorTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/concord/cord04Roles/EditionFoldFloorTest.kt
index 186311dea5..4876b089ef 100644
--- a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/concord/cord04Roles/EditionFoldFloorTest.kt
+++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/concord/cord04Roles/EditionFoldFloorTest.kt
@@ -110,6 +110,84 @@ class EditionFoldFloorTest {
         assertEquals(v3.rumorId, head?.rumorId)
     }
 
+    /**
+     * The cross-epoch case, observed on device: a CORD-06 Refounding re-wraps ONE edition per
+     * entity, so an entity edited *after* the refounding has its successor alone on the new epoch
+     * while the floor edition stays behind on the old one. The offered set therefore never contains
+     * the floor edition — but v3 cites its hash, which is a *stronger* proof the chain connects than
+     * mere presence. Refusing here pins the entity at the floor forever and silently discards the
+     * newer state (a role edit, a channel rename, a banlist entry) on every refold.
+     *
+     * Regression guard: reproduced on device 2026-07-29 (entity e83ee182 pinned at v1 across 21
+     * consecutive refolds while epoch 1 offered v2 citing v1's hash). Mirrors Armada's third
+     * anchor branch, `versions[0] === floor + 1n → bytesEq(lo.prevHash, floorHash)`.
+     */
+    @Test
+    fun advancesWhenOnlyTheConnectingSuccessorIsOffered() {
+        val gaps = mutableListOf<Triple<String, Long, Long>>()
+        val head = EditionFold.foldEntity(listOf(v3), floorAtV2) { e, f, o -> gaps += Triple(e, f, o) }
+
+        assertEquals(v3.rumorId, head?.rumorId, "v3.prev == floor.hash, so the chain connects")
+        assertTrue(gaps.isEmpty(), "a successor citing the floor is not a rollback")
+    }
+
+    /** But a successor that cites something else is still a fork, and still refused. */
+    @Test
+    fun stillRefusesASuccessorThatDoesNotCiteTheFloor() {
+        val forked = edition(3, ByteArray(32) { 0x11 }, """{"member":"a","role_ids":["owner-ish"]}""", rumorId = "id-forked")
+        val gaps = mutableListOf<Triple<String, Long, Long>>()
+        val head = EditionFold.foldEntity(listOf(forked), floorAtV2) { e, f, o -> gaps += Triple(e, f, o) }
+
+        assertEquals(v2.rumorId, head?.rumorId, "an unconnected successor must not be adopted")
+        assertEquals(listOf(Triple(v2.entityIdHex, 2L, 3L)), gaps, "and the refusal must still be reported")
+    }
+
+    // ---- the compaction arm (CORD-06 §3) --------------------------------------
+
+    /**
+     * Behind a Refounding the predecessor is trimmed, so a `prev` that reaches back to the floor
+     * is NOT guaranteed — the compacted head can sit any distance above it. Once the entity is in
+     * the epoch's snapshot the anchor is the version alone, so v3 is adopted over a floor of v1
+     * even though nothing links them. This is the case part 1 alone cannot handle.
+     */
+    @Test
+    fun compactionArmAdoptsTheHighestVersionWithoutAChain() {
+        val gaps = mutableListOf<Triple<String, Long, Long>>()
+        val head =
+            EditionFold.foldEntity(
+                listOf(v3),
+                v1.asFloor(),
+                snapshot = setOf(v3.rumorId),
+            ) { e, f, o -> gaps += Triple(e, f, o) }
+
+        assertEquals(v3.rumorId, head?.rumorId, "version anchors the compaction arm; prev is ignored")
+        assertTrue(gaps.isEmpty(), "a compacted head above the floor is not a gap")
+    }
+
+    /** The floor still holds in the compaction arm: a re-served stale edition cannot downgrade us. */
+    @Test
+    fun compactionArmStillRefusesEverythingBelowTheFloor() {
+        val gaps = mutableListOf<Triple<String, Long, Long>>()
+        val head =
+            EditionFold.foldEntity(
+                listOf(v0, v1),
+                floorAtV2,
+                snapshot = setOf(v0.rumorId, v1.rumorId),
+            ) { e, f, o -> gaps += Triple(e, f, o) }
+
+        assertEquals(v2.rumorId, head?.rumorId, "nothing at or above the floor: keep what we knew")
+        assertEquals(listOf(Triple(v2.entityIdHex, 2L, 1L)), gaps, "and report the refusal")
+    }
+
+    /** An entity this epoch never re-wrapped is not in the snapshot, so it keeps chain-walk semantics. */
+    @Test
+    fun entityAbsentFromTheSnapshotStillChainWalks() {
+        val forked = edition(3, ByteArray(32) { 0x11 }, """{"member":"a","role_ids":["owner-ish"]}""", rumorId = "id-forked")
+        val head = EditionFold.foldEntity(listOf(forked), floorAtV2, snapshot = setOf("some-other-rumor")) { _, _, _ -> }
+
+        assertEquals(v2.rumorId, head?.rumorId, "no snapshot membership -> the chain walk still refuses the fork")
+    }
+
     /** A floor whose known edition we no longer hold still refuses to move down — it adopts nothing. */
     @Test
     fun refusesRollbackWithNoKnownEditionToFallBackOn() {