test(battery): behavioral tests for the keep-alive pause and worker gates

The ping-interval change shipped on reasoning and was overturned by measurement; these tests give the remaining battery changes the same scrutiny where the claims are about OUR code (deterministically testable), instead of leaving them reasoning-only: - NostrClientKeepAliveTest (quartz): a server-closed relay is redialed within one 60s sweep while active; ZERO dials happen while the client is inactive no matter how much time passes; and the sweep provably resumes after connect() — the failure mode a suspend/resume bug in the new isActiveFlow gate would cause. - ScheduledPostWorkGate extracted from AppModules + tests: schedules on the first PENDING post, cancels when the last drains, re-schedules on publishNow-retry, and — the race that motivated the extraction — never emits a spurious cancel from the store flow's empty placeholder before the disk load completes. - CalendarReminderWorker.couldStillFire extracted + tests against LocalCache: future target keeps the chain, past target lets it end, and an unresolved/start-less target keeps it alive so a reminder can't be lost while the target event is still being fetched. The watchdog alarm change (wakeup -> non-wakeup) remains reasoning-only by nature: it asserts Android OS alarm semantics on OEM devices, which no JVM test can exercise; its blast radius is limited to opt-in always-on users with two independent restart layers still active. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016RJ8EAsdkHx5WHHU2eQJ1P
2026-08-09 16:14:40 +00:00 · 2026-07-12 19:32:20 +00:00
parent 4212f38219
commit a760d17eaa
6 changed files with 561 additions and 43 deletions
@@ -87,8 +87,8 @@ import com.vitorpamplona.amethyst.service.relayClient.reqCommand.event.EventFind
 import com.vitorpamplona.amethyst.service.relayClient.reqCommand.user.UserFinderQueryState
 import com.vitorpamplona.amethyst.service.relayClient.speedLogger.RelaySpeedLogger
 import com.vitorpamplona.amethyst.service.safeCacheDir
-import com.vitorpamplona.amethyst.service.scheduledposts.ScheduledPostStatus
 import com.vitorpamplona.amethyst.service.scheduledposts.ScheduledPostStore
+import com.vitorpamplona.amethyst.service.scheduledposts.ScheduledPostWorkGate
 import com.vitorpamplona.amethyst.service.scheduledposts.ScheduledPostWorker
 import com.vitorpamplona.amethyst.service.uploads.blossom.bud10.BlossomServerResolver
 import com.vitorpamplona.amethyst.service.uploads.blossom.bud10.LocalBlossomCacheProbe
@@ -893,30 +893,19 @@ class AppModules(
        notificationDispatcher.start()

        // Keep the scheduled-posts worker (15-min periodic + one-time catch-up)
-        // enqueued exactly while the store holds a PENDING post. An always-on
-        // periodic worker wakes — and often cold-starts — the whole process every
-        // 15 minutes forever, even for users who never schedule a post. The store
-        // is durable (JSON on disk) and the single source of truth, so the worker
-        // is (re-)enqueued from any mutation that produces a PENDING row and
-        // cancelled when the last one drains. Runs independently of the always-on
+        // enqueued exactly while the store holds a PENDING post — see
+        // ScheduledPostWorkGate. Runs independently of the always-on
        // notification setting so scheduled posts still fire when always-on
        // notifications are disabled.
-        applicationIOScope.launch {
-            // Force the initial disk load; the flow's initial value is an empty
-            // list until the store is first touched.
-            scheduledPostStore.list()
-            scheduledPostStore.flow
-                .map { posts -> posts.any { it.status == ScheduledPostStatus.PENDING } }
-                .distinctUntilChanged()
-                .collect { hasPending ->
-                    if (hasPending) {
-                        ScheduledPostWorker.schedule(appContext)
-                        ScheduledPostWorker.scheduleCatchUp(appContext)
-                    } else {
-                        ScheduledPostWorker.cancelPeriodic(appContext)
-                    }
-                }
-        }
+        ScheduledPostWorkGate(
+            store = scheduledPostStore,
+            scope = applicationIOScope,
+            onPendingWork = {
+                ScheduledPostWorker.schedule(appContext)
+                ScheduledPostWorker.scheduleCatchUp(appContext)
+            },
+            onNoPendingWork = { ScheduledPostWorker.cancelPeriodic(appContext) },
+        ).start()

        // "Starting soon" reminders for NIP-52 appointments the user RSVP'd to as
        // ACCEPTED. The 15-min periodic scanner is only scheduled while it can
@@ -76,12 +76,7 @@ class CalendarReminderWorker(
        // multi-account "all logged-in pubkeys" view here, so we accept any RSVP that's
        // present in cache — the alternative (looking only at the foreground account) would
        // silently break notifications for account switching during the lead window.
-        val acceptedRsvps =
-            LocalCache.addressables
-                .filterIntoSet { _, note ->
-                    val e = note.event
-                    e is CalendarRSVPEvent && e.status() == RSVPStatusTag.STATUS.ACCEPTED
-                }.mapNotNull { it.event as? CalendarRSVPEvent }
+        val acceptedRsvps = acceptedRsvpsInCache()

        Log.d(TAG) { "Worker scanning ${acceptedRsvps.size} accepted RSVPs (now=$now, lead=${prefs.leadMinutes()}m)" }

@@ -122,22 +117,10 @@ class CalendarReminderWorker(
        store.forgetBefore(now - PRUNE_AGE_SECONDS)

        // Nothing left that could ever fire → end the periodic chain instead of
-        // waking the process every 15 minutes forever. An RSVP whose target event
-        // hasn't been fetched yet (start unknown) counts as "could still fire" so
-        // the chain survives until the target resolves. The ACCEPTED-RSVP observer
+        // waking the process every 15 minutes forever. The ACCEPTED-RSVP observer
        // in AppModules re-schedules the worker the next time a live session sees
        // an accepted RSVP.
-        val couldStillFire =
-            acceptedRsvps.any { rsvp ->
-                val targetAddress = rsvp.calendarEventAddress() ?: return@any false
-                val start =
-                    LocalCache.addressables
-                        .get(targetAddress)
-                        ?.appointmentView()
-                        ?.startSeconds
-                start == null || start > now
-            }
-        if (!couldStillFire) {
+        if (!couldStillFire(acceptedRsvps, now)) {
            Log.d(TAG) { "No accepted RSVP can still fire; ending periodic chain." }
            cancel(applicationContext)
        }
@@ -152,6 +135,35 @@ class CalendarReminderWorker(
        // more than a day ago; they can't fire again so the entry is pure overhead.
        private const val PRUNE_AGE_SECONDS = 24L * 60L * 60L

+        /** Every ACCEPTED kind-31925 RSVP currently present in LocalCache. */
+        fun acceptedRsvpsInCache(): List<CalendarRSVPEvent> =
+            LocalCache.addressables
+                .filterIntoSet { _, note ->
+                    val e = note.event
+                    e is CalendarRSVPEvent && e.status() == RSVPStatusTag.STATUS.ACCEPTED
+                }.mapNotNull { it.event as? CalendarRSVPEvent }
+
+        /**
+         * True while at least one accepted RSVP could still produce a reminder:
+         * its target event either starts in the future, or hasn't been fetched
+         * yet (start unknown — the chain must survive until the target
+         * resolves). False means the periodic worker has nothing it could ever
+         * notify about and may cancel its own chain.
+         */
+        fun couldStillFire(
+            rsvps: Collection<CalendarRSVPEvent>,
+            now: Long,
+        ): Boolean =
+            rsvps.any { rsvp ->
+                val targetAddress = rsvp.calendarEventAddress() ?: return@any false
+                val start =
+                    LocalCache.addressables
+                        .get(targetAddress)
+                        ?.appointmentView()
+                        ?.startSeconds
+                start == null || start > now
+            }
+
        fun schedule(context: Context) {
            val request =
                PeriodicWorkRequestBuilder<CalendarReminderWorker>(15, TimeUnit.MINUTES)
@@ -0,0 +1,65 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.amethyst.service.scheduledposts
+
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.launch
+
+/**
+ * Keeps [ScheduledPostWorker]'s 15-minute periodic chain enqueued exactly while
+ * the store holds a PENDING post. An unconditionally-scheduled periodic worker
+ * wakes — and often cold-starts — the whole process every 15 minutes forever,
+ * even for users who never schedule a post.
+ *
+ * The store is durable (JSON on disk) and the single source of truth, so
+ * [onPendingWork] fires from any mutation that produces a PENDING row (add,
+ * publishNow, releaseClaim) and [onNoPendingWork] when the last one drains
+ * (markSent/markFailed/cancel/removeForAccount).
+ *
+ * [start] forces the store's initial disk load BEFORE collecting: the flow's
+ * initial value is an empty list until the store is first touched, and acting
+ * on that placeholder would cancel scheduled work that a pending post still
+ * needs.
+ */
+class ScheduledPostWorkGate(
+    private val store: ScheduledPostStore,
+    private val scope: CoroutineScope,
+    private val onPendingWork: () -> Unit,
+    private val onNoPendingWork: () -> Unit,
+) {
+    fun start(): Job =
+        scope.launch {
+            store.list()
+            store.flow
+                .map { posts -> posts.any { it.status == ScheduledPostStatus.PENDING } }
+                .distinctUntilChanged()
+                .collect { hasPending ->
+                    if (hasPending) {
+                        onPendingWork()
+                    } else {
+                        onNoPendingWork()
+                    }
+                }
+        }
+}
@@ -0,0 +1,149 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.amethyst.calendar
+
+import com.vitorpamplona.amethyst.model.LocalCache
+import com.vitorpamplona.amethyst.service.calendar.CalendarReminderWorker
+import com.vitorpamplona.quartz.nip52Calendar.appt.time.CalendarTimeSlotEvent
+import com.vitorpamplona.quartz.nip52Calendar.rsvp.CalendarRSVPEvent
+import org.junit.Assert.assertFalse
+import org.junit.Assert.assertTrue
+import org.junit.Test
+
+/**
+ * The predicate that decides whether the CalendarReminderWorker's periodic
+ * chain may cancel itself. Getting it wrong in either direction is a bug:
+ * a false "could fire" keeps waking the process forever; a false "can't fire"
+ * silently kills a reminder the user RSVP'd to.
+ */
+class CalendarReminderCandidatesTest {
+    private val now = 2_000_000_000L
+
+    // Unique per-test-class identities so the shared LocalCache singleton
+    // doesn't collide with other suites running in the same JVM.
+    private val organizer = "b0b0000000000000000000000000000000000000000000000000000000000001"
+    private val attendee = "a11ce00000000000000000000000000000000000000000000000000000000002"
+
+    private var eventSeq = 0
+
+    private fun timeSlot(
+        dTag: String,
+        startSeconds: Long?,
+    ) = CalendarTimeSlotEvent(
+        id = "c0ffee%058d".format(++eventSeq),
+        pubKey = organizer,
+        createdAt = now - 1000,
+        tags =
+            if (startSeconds != null) {
+                arrayOf(arrayOf("d", dTag), arrayOf("start", startSeconds.toString()))
+            } else {
+                arrayOf(arrayOf("d", dTag))
+            },
+        content = "",
+        sig = "sig",
+    )
+
+    private fun rsvp(
+        dTag: String,
+        targetDTag: String?,
+        status: String = "accepted",
+    ) = CalendarRSVPEvent(
+        id = "faced%059d".format(++eventSeq),
+        pubKey = attendee,
+        createdAt = now - 900,
+        tags =
+            buildList {
+                add(arrayOf("d", dTag))
+                add(arrayOf("status", status))
+                if (targetDTag != null) {
+                    add(arrayOf("a", "${CalendarTimeSlotEvent.KIND}:$organizer:$targetDTag"))
+                }
+            }.toTypedArray(),
+        content = "",
+        sig = "sig",
+    )
+
+    @Test
+    fun acceptedRsvpsInCache_returnsAcceptedAndSkipsDeclined() {
+        val accepted = rsvp("cand-accepted", "cand-target-1")
+        val declined = rsvp("cand-declined", "cand-target-2", status = "declined")
+        LocalCache.justConsume(accepted, null, true)
+        LocalCache.justConsume(declined, null, true)
+
+        val found = CalendarReminderWorker.acceptedRsvpsInCache()
+        assertTrue("accepted RSVP must be found", found.any { it.dTag() == "cand-accepted" })
+        assertFalse("declined RSVP must be skipped", found.any { it.dTag() == "cand-declined" })
+    }
+
+    @Test
+    fun futureTarget_couldStillFire() {
+        val slot = timeSlot("cand-future", startSeconds = now + 3600)
+        val r = rsvp("cand-rsvp-future", "cand-future")
+        LocalCache.justConsume(slot, null, true)
+        LocalCache.justConsume(r, null, true)
+
+        assertTrue(CalendarReminderWorker.couldStillFire(listOf(r), now))
+    }
+
+    @Test
+    fun pastTarget_cannotFireAnymore() {
+        val slot = timeSlot("cand-past", startSeconds = now - 3600)
+        val r = rsvp("cand-rsvp-past", "cand-past")
+        LocalCache.justConsume(slot, null, true)
+        LocalCache.justConsume(r, null, true)
+
+        assertFalse(CalendarReminderWorker.couldStillFire(listOf(r), now))
+    }
+
+    @Test
+    fun unresolvedTarget_keepsTheChainAlive() {
+        // The RSVP points at an event the cache hasn't fetched yet: the start
+        // is unknown, so the worker must NOT cancel its chain.
+        val r = rsvp("cand-rsvp-unresolved", "cand-never-fetched")
+        LocalCache.justConsume(r, null, true)
+
+        assertTrue(CalendarReminderWorker.couldStillFire(listOf(r), now))
+    }
+
+    @Test
+    fun targetWithoutStart_keepsTheChainAlive() {
+        // Target resolved but carries no start tag: still unknown, keep alive.
+        val slot = timeSlot("cand-no-start", startSeconds = null)
+        val r = rsvp("cand-rsvp-no-start", "cand-no-start")
+        LocalCache.justConsume(slot, null, true)
+        LocalCache.justConsume(r, null, true)
+
+        assertTrue(CalendarReminderWorker.couldStillFire(listOf(r), now))
+    }
+
+    @Test
+    fun rsvpWithoutTargetAddress_doesNotKeepTheChainAlive() {
+        val r = rsvp("cand-rsvp-no-a-tag", targetDTag = null)
+        LocalCache.justConsume(r, null, true)
+
+        assertFalse(CalendarReminderWorker.couldStillFire(listOf(r), now))
+    }
+
+    @Test
+    fun emptyCache_doesNotKeepTheChainAlive() {
+        assertFalse(CalendarReminderWorker.couldStillFire(emptyList(), now))
+    }
+}
@@ -0,0 +1,149 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.amethyst.service.scheduledposts
+
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.runCurrent
+import kotlinx.coroutines.test.runTest
+import org.junit.Assert.assertEquals
+import org.junit.Before
+import org.junit.Rule
+import org.junit.Test
+import org.junit.rules.TemporaryFolder
+import java.io.File
+
+/**
+ * The gate must keep the periodic worker enqueued exactly while a PENDING
+ * post exists — and, critically, must never act on the store flow's empty
+ * placeholder value before the disk load completes (which would cancel
+ * scheduled work an existing pending post still needs).
+ */
+@OptIn(ExperimentalCoroutinesApi::class)
+class ScheduledPostWorkGateTest {
+    @get:Rule
+    val temp = TemporaryFolder()
+
+    private lateinit var file: File
+
+    /** Chronological record of gate decisions: true = schedule, false = cancel. */
+    private val decisions = mutableListOf<Boolean>()
+
+    @Before
+    fun setUp() {
+        file = File(temp.root, "scheduled_posts.json")
+        decisions.clear()
+    }
+
+    private fun newStore() = ScheduledPostStore(file)
+
+    private fun TestScope.startGate(store: ScheduledPostStore) =
+        ScheduledPostWorkGate(
+            store = store,
+            scope = backgroundScope,
+            onPendingWork = { decisions.add(true) },
+            onNoPendingWork = { decisions.add(false) },
+        ).start()
+
+    private fun samplePost(
+        id: String = "id-1",
+        publishAtSec: Long = 4_000_000_000,
+    ) = ScheduledPost(
+        id = id,
+        accountPubkey = "pk1",
+        signedEventJson = "{}",
+        relayUrls = listOf("wss://relay.example/"),
+        extraEventsJson = emptyList(),
+        publishAtSec = publishAtSec,
+        createdAtSec = 500,
+    )
+
+    @Test
+    fun emptyStore_cancelsOnceAndOnlyOnce() =
+        runTest {
+            startGate(newStore())
+            runCurrent()
+            assertEquals(listOf(false), decisions)
+        }
+
+    @Test
+    fun pendingPostOnDisk_schedulesWithoutSpuriousCancelFirst() =
+        runTest {
+            // Persist a pending post in a previous "process".
+            runBlocking { newStore().add(samplePost()) }
+
+            // Fresh store (flow starts as the empty placeholder). The gate must
+            // load from disk before collecting: the FIRST decision must be
+            // "schedule", never a cancel from the placeholder value.
+            startGate(newStore())
+            runCurrent()
+            assertEquals(listOf(true), decisions)
+        }
+
+    @Test
+    fun addThenDrainThenAddAgain_togglesTheWorker() =
+        runTest {
+            val store = newStore()
+            startGate(store)
+            runCurrent()
+            assertEquals(listOf(false), decisions)
+
+            store.add(samplePost(id = "a"))
+            runCurrent()
+            assertEquals(listOf(false, true), decisions)
+
+            // A second pending post must not re-fire (distinctUntilChanged).
+            store.add(samplePost(id = "b"))
+            runCurrent()
+            assertEquals(listOf(false, true), decisions)
+
+            // Draining both pending posts cancels the periodic chain.
+            store.markSent("a")
+            runCurrent()
+            assertEquals(listOf(false, true), decisions)
+            store.markFailed("b", "boom")
+            runCurrent()
+            assertEquals(listOf(false, true, false), decisions)
+
+            // A new post re-schedules.
+            store.add(samplePost(id = "c"))
+            runCurrent()
+            assertEquals(listOf(false, true, false, true), decisions)
+        }
+
+    @Test
+    fun publishNowOnFailedPost_reschedulesTheWorker() =
+        runTest {
+            val store = newStore()
+            store.add(samplePost(id = "a"))
+            startGate(store)
+            runCurrent()
+            store.markFailed("a", "relay down")
+            runCurrent()
+            assertEquals(listOf(true, false), decisions)
+
+            // Retry flips the post back to PENDING; the worker must come back.
+            store.publishNow("a")
+            runCurrent()
+            assertEquals(listOf(true, false, true), decisions)
+        }
+}
@@ -0,0 +1,154 @@
+/*
+ * Copyright (c) 2025 Vitor Pamplona
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of
+ * this software and associated documentation files (the "Software"), to deal in
+ * the Software without restriction, including without limitation the rights to use,
+ * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
+ * Software, and to permit persons to whom the Software is furnished to do so,
+ * subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
+ * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+package com.vitorpamplona.quartz.nip01Core.relay.client
+
+import com.vitorpamplona.quartz.nip01Core.relay.client.single.basic.FakeWebsocketBuilder
+import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.advanceTimeBy
+import kotlinx.coroutines.test.runCurrent
+import kotlinx.coroutines.test.runTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+/**
+ * Behavioral contract of the keep-alive sweep after it was changed to suspend
+ * on the client's active-state flow instead of ticking unconditionally:
+ *
+ *  1. while ACTIVE, a relay closed by the server is redialed within one 60s
+ *     sweep interval (once the per-relay backoff gate opens);
+ *  2. while INACTIVE (host called [NostrClient.disconnect], i.e. the app went
+ *     to the background), no amount of elapsed time produces a dial;
+ *  3. after [NostrClient.connect] reactivates the client, redialing works
+ *     again — a bug in the suspend/resume logic would leave every
+ *     server-dropped relay disconnected forever.
+ *
+ * Coroutine timers (keep-alive 60s, reconnect debounce 200ms) run on the test
+ * scheduler's virtual clock; the per-relay backoff gate compares real wall
+ * seconds (integer granularity), hence the short real sleeps before each
+ * expected redial.
+ *
+ * Harness note: every socket the client dials must eventually be driven to
+ * onOpen/onClosed — a [FakeWebsocketBuilder] socket that is never completed
+ * leaves the relay in "connecting" forever and blocks all later dials, which
+ * is exactly what a production dial never does. [settleDisconnected] flushes
+ * pending debounced reconnects and completes any socket they may open.
+ */
+@OptIn(ExperimentalCoroutinesApi::class)
+class NostrClientKeepAliveTest {
+    private val url = NormalizedRelayUrl("wss://keepalive.example.com")
+
+    /**
+     * Flushes pending sub-second timers (the 200ms reconnect debounce) and,
+     * if a flush dialed a new socket, completes its lifecycle so the relay
+     * ends DISCONNECTED with no pending timers besides the keep-alive sweep.
+     */
+    private fun TestScope.settleDisconnected(builder: FakeWebsocketBuilder) {
+        repeat(4) {
+            val before = builder.connectAttempts
+            advanceTimeBy(500)
+            runCurrent()
+            if (builder.connectAttempts == before) return
+            builder.lastListener.onOpen(50, false)
+            builder.lastListener.onClosed(1000, "test settle")
+        }
+    }
+
+    @Test
+    fun keepAliveSweepPausesWhileInactiveAndResumesAfterReconnect() =
+        runTest {
+            val builder = FakeWebsocketBuilder()
+            val client = NostrClient(builder, this)
+            try {
+                // Flush refreshConnection's initial emission against the
+                // still-empty pool so it can't dial later.
+                advanceTimeBy(500)
+                runCurrent()
+
+                // Subscribing dials the relay immediately.
+                client.subscribe("sub", mapOf(url to listOf(Filter())))
+                assertEquals(1, builder.connectAttempts)
+
+                // Server closes the established connection. While ACTIVE the
+                // client must redial within one keep-alive interval.
+                builder.lastListener.onOpen(50, false)
+                builder.lastListener.onClosed(1000, "server closed")
+                settleDisconnected(builder)
+                val beforeActiveSweep = builder.connectAttempts
+                Thread.sleep(3_500)
+                advanceTimeBy(61_000)
+                runCurrent()
+                assertTrue(
+                    builder.connectAttempts > beforeActiveSweep,
+                    "active client should redial a server-closed relay within one sweep, " +
+                        "got ${builder.connectAttempts} attempts (baseline $beforeActiveSweep)",
+                )
+
+                // Leave the relay cleanly disconnected, then deactivate.
+                builder.lastListener.onOpen(50, false)
+                builder.lastListener.onClosed(1000, "server closed")
+                settleDisconnected(builder)
+                client.disconnect()
+                val whileInactiveBaseline = builder.connectAttempts
+
+                // The relay's backoff gate is reset by disconnect(), so any
+                // stray sweep tick WOULD dial — this fails if the sweep keeps
+                // running (or anything else dials) while inactive.
+                Thread.sleep(3_500)
+                advanceTimeBy(30 * 60_000)
+                runCurrent()
+                assertEquals(
+                    whileInactiveBaseline,
+                    builder.connectAttempts,
+                    "no dial may happen while the client is inactive",
+                )
+
+                // Reactivation dials the pool immediately...
+                client.connect()
+                runCurrent()
+                assertEquals(
+                    whileInactiveBaseline + 1,
+                    builder.connectAttempts,
+                    "connect() should redial the pool immediately",
+                )
+
+                // ...and after the pause a server-closed relay must again be
+                // redialed within one sweep — fails if the sweep never
+                // resumes after the inactive stretch.
+                builder.lastListener.onOpen(50, false)
+                builder.lastListener.onClosed(1000, "server closed")
+                settleDisconnected(builder)
+                val beforeResumedSweep = builder.connectAttempts
+                Thread.sleep(3_500)
+                advanceTimeBy(61_000)
+                runCurrent()
+                assertTrue(
+                    builder.connectAttempts > beforeResumedSweep,
+                    "sweep did not resume after connect(); a server-dropped relay would stay disconnected forever",
+                )
+            } finally {
+                client.close()
+            }
+        }
+}