fix(dm): realign the per-relay download window when DMs are pruned

Memory pruning drops DM messages out of the cache but left the per-relay
paging cursors untouched, so a relay still claimed to have delivered the
dropped band (reachedUntil deep, or done) and the demand-driven loader
never re-requested it — a silent hole until app restart.

- Prune NIP-17 too: pruneMessagesToTheLatestOnly now reaps both NIP-04
  (PrivateDmEvent) and NIP-17 (WrappedEvent rumors) on one merged top-N
  cut, so a conversation is cut at a single time point (no NIP-04-without
  -NIP-17 holes). NIP-17 is the actual memory-pressure driver.
- HostStub carries the host's createdAt, so a decrypted rumor self-
  describes its outer gift-wrap time (the time the cursor pages by; the
  rumor's own time is the message time, not the wrap time).
- RelayLoadingCursors.rewindTo() pulls a relay's reached cursor up past
  the pruned band, clears done, and un-arms it (demand-driven re-fetch);
  advance() now resumes from the rewound reached point instead of the
  floor.
- LocalCache.pruneOldMessages accumulates the newest pruned created_at
  per relay (outer-wrap time for gift wraps, event time for NIP-04),
  filtered below each cursor's floor, then rewinds giftWrapHistory +
  rooms-list nip04History (account-wide) and the per-conversation
  nip04History.

The gift-wrap window is account-global, so pruning one room rewinds the
shared sweep; the interference is bounded (already-held wraps short-
circuit in consumeRegularEvent, re-fetch is demand-gated).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

amethyst/src/main/java/com/vitorpamplona/amethyst/model/LocalCache.kt

@@ -2645,20 +2645,58 @@ object LocalCache : ILocalCache, ICacheProvider {
         }
 
         chatroomList.forEach { userHex, room ->
+            // History floors are pinned per scope on first advance; null means that window never paged
+            // history, so its cursors hold no position to misalign and nothing needs rewinding. Only the
+            // bands strictly BELOW a floor are this window's responsibility — a pruned message newer than
+            // the floor is the always-on live tail's concern, and rewinding history for it would needlessly
+            // re-page (and, for a busy room straddling the floor, mis-set the boundary). Hence the per-floor
+            // filter when accumulating below.
+            val giftWrapFloor = room.giftWrapHistory.floor
+            val accountNip04Floor = room.nip04History.floor
+
             room.rooms.map { key, chatroom ->
                 val toBeRemoved = chatroom.pruneMessagesToTheLatestOnly()
 
                 val childrenToBeRemoved = mutableListOf<Note>()
 
-                toBeRemoved.forEach {
-                    childrenToBeRemoved.addAll(removeIfWrap(it))
-                    unlinkAndRemove(it)
+                // Newest pruned `created_at` per relay, in each window's cursor space, capped at < floor.
+                // Gift wraps page by the OUTER wrap time (from the rumor's host stub); NIP-04 by the event's
+                // own time, and a kind:4 belongs to BOTH the account (rooms-list) and per-conversation cursor.
+                val giftWrapPruned = HashMap<NormalizedRelayUrl, Long>()
+                val accountNip04Pruned = HashMap<NormalizedRelayUrl, Long>()
+                val roomNip04Pruned = HashMap<NormalizedRelayUrl, Long>()
+                // chatroom.nip04History is lazy — only touch (allocate) it when this room actually drops a
+                // kind:4 message, so rooms that never paged conversation history pay nothing.
+                val roomNip04Floor = if (toBeRemoved.any { it.event is PrivateDmEvent }) chatroom.nip04History.floor else null
 
-                    childrenToBeRemoved.addAll(it.clearChildLinks())
+                toBeRemoved.forEach { note ->
+                    when (val ev = note.event) {
+                        is WrappedEvent ->
+                            if (giftWrapFloor != null) {
+                                val outerUntil = ev.host?.createdAt ?: ev.createdAt
+                                if (outerUntil < giftWrapFloor) note.relays.forEach { giftWrapPruned.merge(it, outerUntil, ::maxOf) }
+                            }
+                        is PrivateDmEvent -> {
+                            val until = ev.createdAt
+                            if (accountNip04Floor != null && until < accountNip04Floor) note.relays.forEach { accountNip04Pruned.merge(it, until, ::maxOf) }
+                            if (roomNip04Floor != null && until < roomNip04Floor) note.relays.forEach { roomNip04Pruned.merge(it, until, ::maxOf) }
+                        }
+                    }
+
+                    childrenToBeRemoved.addAll(removeIfWrap(note))
+                    unlinkAndRemove(note)
+
+                    childrenToBeRemoved.addAll(note.clearChildLinks())
                 }
 
                 unlinkAndRemove(childrenToBeRemoved)
 
+                // Realign the windows so a relay that already paged past (or `done` below) the dropped band
+                // re-requests it on the next demand-advance instead of skipping the hole.
+                if (giftWrapPruned.isNotEmpty()) room.giftWrapHistory.rewindTo(giftWrapPruned)
+                if (accountNip04Pruned.isNotEmpty()) room.nip04History.rewindTo(accountNip04Pruned)
+                if (roomNip04Pruned.isNotEmpty()) chatroom.nip04History.rewindTo(roomNip04Pruned)
+
                 if (toBeRemoved.size > 1) {
                     println(
                         "PRUNE: ${toBeRemoved.size} private messages from $userHex to ${key.users.joinToString()} removed. ${chatroom.messages.size} kept",

amethyst/src/test/java/com/vitorpamplona/amethyst/service/relayClient/eoseManagers/RelayLoadingCursorsTest.kt

@@ -106,4 +106,101 @@ class RelayLoadingCursorsTest {
         val cursors = RelayLoadingCursors()
         assertEquals(null, cursors.deepestReached(emptyList(), start))
     }
+
+    // ── rewindTo: realign the window after the cache prunes messages out of it ──
+
+    @Test
+    fun rewindReopensThePrunedBandAndResumesFromItOnNextAdvance() {
+        val cursors = RelayLoadingCursors()
+        cursors.floor = start
+
+        // page deep: floor 1000 → reached 200
+        cursors.advance(relayA, start)
+        cursors.onEvent(relayA, 900)
+        cursors.onEvent(relayA, 200)
+        cursors.onEose(relayA)
+        assertEquals(200L, cursors.reachedUntilFor(relayA, start))
+
+        // prune drops everything older than 700 (newest pruned = 700)
+        cursors.rewindTo(mapOf(relayA to 700L))
+
+        // reached pulled up to just above the pruned band, not done, and un-armed (demand-driven)
+        assertEquals(701L, cursors.reachedUntilFor(relayA, start))
+        assertFalse(cursors.isDone(relayA))
+        assertEquals(emptyList<Any>(), cursors.armedRelays(listOf(relayA)))
+
+        // the next advance resumes at the boundary and re-requests the pruned band (until = 700),
+        // NOT from the floor (which would re-stream the still-held tail above 700)
+        assertTrue(cursors.advance(relayA, start))
+        assertEquals(700L, cursors.requestedUntilFor(relayA))
+    }
+
+    @Test
+    fun rewindClearsDoneSoAnExhaustedRelayCanReFetch() {
+        val cursors = RelayLoadingCursors()
+        cursors.floor = start
+
+        cursors.advance(relayA, start)
+        cursors.onEvent(relayA, 300)
+        cursors.onEose(relayA) // reached 300
+        cursors.advance(relayA, start)
+        cursors.onEose(relayA) // empty page → done
+        assertTrue(cursors.isDone(relayA))
+
+        cursors.rewindTo(mapOf(relayA to 500L))
+
+        assertFalse("a pruned relay must be re-fetchable even after it reached the bottom", cursors.isDone(relayA))
+        assertEquals(501L, cursors.reachedUntilFor(relayA, start))
+        assertTrue(cursors.advance(relayA, start))
+        assertEquals(500L, cursors.requestedUntilFor(relayA))
+    }
+
+    @Test
+    fun rewindNeverClimbsAboveTheFloor() {
+        val cursors = RelayLoadingCursors()
+        cursors.floor = start
+
+        cursors.advance(relayA, start)
+        cursors.onEvent(relayA, 300)
+        cursors.onEose(relayA) // reached 300
+
+        // a boundary at/above the floor clamps to the floor (history lives strictly below it)
+        cursors.rewindTo(mapOf(relayA to start))
+        assertEquals(start, cursors.reachedUntilFor(relayA, start))
+    }
+
+    @Test
+    fun rewindSkipsRelaysWithoutACursorOrShallowerThanThePrunedBand() {
+        val cursors = RelayLoadingCursors()
+        cursors.floor = start
+
+        // A delivered to 200; B never paged
+        cursors.advance(relayA, start)
+        cursors.onEvent(relayA, 200)
+        cursors.onEose(relayA)
+
+        // B has no cursor (never paged) → skipped, no entry minted; A's reach (200) is already shallower
+        // than a boundary of 150 (target 151), so it needs no rewind either.
+        cursors.rewindTo(mapOf(relayB to 500L, relayA to 150L))
+
+        // B: untouched (still at the floor, unarmed)
+        assertEquals(start, cursors.reachedUntilFor(relayB, start))
+        assertEquals(emptyList<Any>(), cursors.armedRelays(listOf(relayB)))
+        // A: unchanged
+        assertEquals(200L, cursors.reachedUntilFor(relayA, start))
+    }
+
+    @Test
+    fun rewindIsANoOpWhenTheWindowNeverPagedHistory() {
+        val cursors = RelayLoadingCursors()
+        // floor is null (never advanced any history page)
+        cursors.advance(relayA, start)
+        cursors.onEvent(relayA, 200)
+        cursors.onEose(relayA)
+
+        cursors.rewindTo(mapOf(relayA to 150L))
+
+        // unchanged: with no pinned floor there is no history window to realign
+        assertEquals(200L, cursors.reachedUntilFor(relayA, start))
+    }
 }

commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/model/privateChats/Chatroom.kt

@@ -33,6 +33,7 @@ import com.vitorpamplona.quartz.nip01Core.core.HexKey
 import com.vitorpamplona.quartz.nip01Core.relay.client.paging.RelayLoadingCursors
 import com.vitorpamplona.quartz.nip04Dm.messages.PrivateDmEvent
 import com.vitorpamplona.quartz.nip14Subject.subject
+import com.vitorpamplona.quartz.nip59Giftwrap.WrappedEvent
 import com.vitorpamplona.quartz.utils.TimeUtils
 import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.flow.MutableSharedFlow
@@ -145,7 +146,11 @@ class Chatroom : NotesGatherer {
             } else {
                 // Old messages, keep the last one.
                 sorted.take(1).toSet()
-            } + sorted.filter { it.flowSet?.isInUse() ?: false } + sorted.filter { it.event !is PrivateDmEvent }
+            } + sorted.filter { it.flowSet?.isInUse() ?: false } + sorted.filter { it.event !is PrivateDmEvent && it.event !is WrappedEvent }
+        // Both DM protocols are pruned by the recency rule above: NIP-04 (PrivateDmEvent) and NIP-17
+        // (WrappedEvent rumors — ChatMessageEvent / file headers). Anything else that ever lands in a
+        // room is kept. The caller realigns the per-relay download window for the dropped messages so
+        // they can be paged again later (see LocalCache.pruneOldMessages + RelayLoadingCursors.rewindTo).
 
         val toRemove = messages.minus(toKeep)
         messages = toKeep

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/relay/client/paging/RelayLoadingCursors.kt

@@ -103,11 +103,18 @@ class RelayLoadingCursors {
     ): Boolean {
         val c = cursor(relay)
         if (c.done) return false
+        val reached = c.reachedUntil
         c.requestedUntil =
-            if (c.requestedUntil == null) {
-                start
-            } else {
-                (c.reachedUntil ?: start) - 1
+            when {
+                // Resume just below the oldest event already delivered. Covers both normal page-to-page
+                // advance and a post-[rewindTo] resume (which un-arms the relay — requestedUntil back to
+                // null — but keeps the rewound reached point, so the next page picks up at the boundary
+                // instead of restarting at the floor and re-streaming the still-held tail).
+                reached != null -> reached - 1
+                // Very first page for this relay (nothing delivered, nothing requested yet).
+                c.requestedUntil == null -> start
+                // Armed but still mid-page (no EOSE yet) — keep asking from the same top.
+                else -> start
             }
         c.pageCount = 0
         c.pageOldest = Long.MAX_VALUE
@@ -147,6 +154,45 @@ class RelayLoadingCursors {
         }
     }
 
+    /**
+     * Realigns the window after the cache prunes messages out of it: for each `relay → newestPrunedUntil`
+     * entry, rewinds that relay so it no longer claims to hold anything at or below [newestPrunedUntil]
+     * (the newest cursor-space `created_at` among the messages pruned from that relay — for gift wraps the
+     * **outer-wrap** time, recovered from the rumor's [host][com.vitorpamplona.quartz.nip59Giftwrap.HostStub]).
+     *
+     * Without this, a relay that already paged past the pruned band — or reached `done` — would never
+     * re-request the dropped messages: its [reachedUntil] still points below them, so the next [advance]
+     * starts even older and skips the hole entirely.
+     *
+     * The rewind pulls [reachedUntil] back up to just above [newestPrunedUntil] (so the next page's
+     * `until` re-includes it), clears [done] (there *is* older data to re-fetch again), and un-arms the
+     * relay (requested cursor back to null) so paging stays demand-driven — the dropped band comes back
+     * only when the on-screen marker advances the relay again, not eagerly on the next re-subscribe.
+     *
+     * Bounds:
+     *  - A relay with no cursor yet (never paged) is skipped — there is no window position to misalign.
+     *  - The rewind never moves [reachedUntil] above the pinned [floor] (history lives strictly below it;
+     *    a pruned message newer than the floor is the live tail's concern, not this window's).
+     *  - A relay whose reached point is already shallower than the pruned band needs no rewind.
+     */
+    fun rewindTo(newestPrunedUntil: Map<NormalizedRelayUrl, Long>) {
+        val floorAt = floor ?: return
+        newestPrunedUntil.forEach { (relay, prunedUntil) ->
+            val c = cursors.get(relay) ?: return@forEach
+            val reached = c.reachedUntil ?: return@forEach
+            // Re-include the newest pruned event: the next page asks `until = reached - 1`, so reached must
+            // sit one tick above it. Never climb above the floor.
+            val target = minOf(prunedUntil + 1, floorAt)
+            if (reached < target) {
+                c.reachedUntil = target
+                c.requestedUntil = null
+                c.done = false
+                c.pageCount = 0
+                c.pageOldest = Long.MAX_VALUE
+            }
+        }
+    }
+
     /** Relays from [all] that have been armed (advanced at least once) and are not yet [isDone]. */
     fun armedRelays(all: Collection<NormalizedRelayUrl>): List<NormalizedRelayUrl> =
         all.filter {

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip59Giftwrap/HostStub.kt

@@ -22,8 +22,19 @@ package com.vitorpamplona.quartz.nip59Giftwrap
 
 import com.vitorpamplona.quartz.nip01Core.core.HexKey
 
+/**
+ * A lightweight reference to the host event a [WrappedEvent] was extracted from — kept on the inner
+ * event so callers can broadcast / delete / locate the outer wrap without holding the full event.
+ *
+ * [createdAt] is the host's own `created_at` (e.g. the kind:1059 gift-wrap timestamp, randomized per
+ * NIP-59), carried here so a decrypted rumor self-describes its outer-wrap time. The history pager
+ * cursors page gift wraps by that outer time, so the prune path uses it to realign the per-relay
+ * download window when a wrapped message is pruned (the chatroom only keeps the inner rumor, whose
+ * `created_at` is the real message time, not the wrap time).
+ */
 class HostStub(
     val id: HexKey,
     val pubKey: HexKey,
     val kind: Int,
+    val createdAt: Long,
 )

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip59Giftwrap/seals/SealedRumorEvent.kt

@@ -70,7 +70,7 @@ class SealedRumorEvent(
 
         val event = rumor.mergeWith(this)
         if (event is WrappedEvent) {
-            event.host = host ?: HostStub(this.id, this.pubKey, this.kind)
+            event.host = host ?: HostStub(this.id, this.pubKey, this.kind, this.createdAt)
         }
         innerEventId = event.id
 

quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip59Giftwrap/wraps/GiftWrapEvent.kt

@@ -74,7 +74,7 @@ open class GiftWrapEvent(
         val gift = fromJson(giftStr)
 
         if (gift is WrappedEvent) {
-            gift.host = HostStub(this.id, this.pubKey, this.kind)
+            gift.host = HostStub(this.id, this.pubKey, this.kind, this.createdAt)
         }
         innerEventId = gift.id