feat(desktop,cli): route WoT kind-3 fetch through OutboxDispatcher (NIP-65)

Phase 3 of the outbox refactor (PR #3483, per Vitor's directive). The
WoT service's kind-3 seeding on Desktop and the `amy wot sync` verb now
go through OutboxDispatcher — index relays discover each author's
kind-10002 write relays, then per-outbox-relay REQs fetch kind-3.

Changes:

  Desktop:
    - DesktopRelaySubscriptionsCoordinator gains an inner
      OutboxCacheGateway that bridges DesktopLocalCache
      (cachedAdvertisedRelayList / consume) to OutboxDispatcher.
    - New suspend loadKind3ViaOutbox(pubkeys) method returns the
      dispatcher's Result for observability.
    - Main.kt WoT-seed effect now:
        1. gates on wotService.isDisabled to preserve MAX_FOLLOWS
           guardrail (fix 2 from Phase 1)
        2. calls loadKind3ViaOutbox instead of the direct
           loadKind3Batched on index relays
        3. keeps the 2s markReady safety net for cold-start UX
    - clear() now also clears outboxDispatcher's dedup markers.

  amy:
    - WotCommand.sync rewritten to construct an OutboxDispatcher, buffer
      events in the gateway, and persist to ctx.store after fetch
      returns (store.insert is suspending; can't call from non-suspend
      gateway callbacks).
    - --json output additively gains kind10002_received,
      outbox_covered_authors, fallback_authors, persisted keys.
    - --timeout N still supported; now maps to overallTimeoutMs.

Not in this commit (deferred to a follow-up on same PR if reviewers
want it):
  - Routing stranger-avatar kind-0 fetch through the outbox path
    (MetadataPreloader wiring is more invasive; keeps this diff focused
    on the primary WoT concern).

Plan: commons/plans/2026-07-06-fix-wot-outbox-model-and-review-fixes-plan.md

cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/WotCommand.kt

@@ -24,14 +24,18 @@ import com.vitorpamplona.amethyst.cli.Args
 import com.vitorpamplona.amethyst.cli.Context
 import com.vitorpamplona.amethyst.cli.DataDir
 import com.vitorpamplona.amethyst.cli.Output
+import com.vitorpamplona.amethyst.commons.wot.OutboxCacheGateway
+import com.vitorpamplona.amethyst.commons.wot.OutboxDispatcher
 import com.vitorpamplona.amethyst.commons.wot.WoTService
+import com.vitorpamplona.quartz.nip01Core.core.Event
 import com.vitorpamplona.quartz.nip01Core.core.HexKey
-import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
-import com.vitorpamplona.quartz.nip02FollowList.ContactListEvent
+import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.cancel
+import java.util.Collections
 
 /**
  * `amy wot <get|list|sync>` — Web-of-Trust score queries.
@@ -113,41 +117,88 @@ object WotCommand {
         rest: Array<String>,
     ): Int {
         val args = Args(rest)
-        val timeoutMs = args.flag("timeout")?.toLongOrNull()?.times(1000) ?: 5_000L
+        // Overall timeout; per-relay budget is set by OutboxDispatcher's
+        // default (4s). `--timeout N` overrides the overall cap.
+        val overallTimeoutMs = args.flag("timeout")?.toLongOrNull()?.times(1000) ?: 8_000L
         Context.open(dataDir).use { ctx ->
             ctx.prepare()
             val self = ctx.identity.pubKeyHex
             val myKind3 = ctx.contactsOf(self)
             val follows =
-                myKind3?.verifiedFollowKeySet()?.toList()
+                myKind3?.verifiedFollowKeySet()?.toSet()
                     ?: return Output.error("no_follows", "no kind-3 in local store; run `amy follow` first")
             if (follows.isEmpty()) {
                 Output.emit(mapOf("synced" to 0, "detail" to "empty follow set"))
                 return 0
             }
-            // Index relays — shared with the Desktop app via
-            // `java.util.prefs`. Falls back to
-            // `PreferencesIndexRelays.DEFAULT_INDEX_RELAYS` when the
-            // user hasn't configured anything, so this is never empty
-            // in practice.
             val relays = ctx.indexRelays()
             if (relays.isEmpty()) return Output.error("no_relays", "no index relays configured")
 
-            // Chunk authors into ≤100 per Filter for relays with per-filter caps.
-            val filters =
-                follows.chunked(100).map { chunk ->
-                    Filter(
-                        kinds = listOf(ContactListEvent.KIND),
-                        authors = chunk,
-                        limit = chunk.size,
+            val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
+            try {
+                // Buffer discovered events; persist synchronously after
+                // the fetch. `store.insert` is suspending so we can't call
+                // it from the non-suspending gateway callbacks. This also
+                // keeps `insert` errors surfaceable in a single log line
+                // rather than swallowed into a race.
+                val buffered = Collections.synchronizedList(mutableListOf<Event>())
+                val gateway =
+                    object : OutboxCacheGateway {
+                        override fun cachedOutbox(pubkey: HexKey): AdvertisedRelayListEvent? =
+                            // Amy's store lookup is suspending; can't do
+                            // it here. The dispatcher then falls through
+                            // to Phase 1 discovery for every author, which
+                            // matches the old `amy wot sync` behaviour of
+                            // always re-asking. A future optimisation
+                            // could pre-populate a `Map<HexKey,
+                            // AdvertisedRelayListEvent>` before dispatch.
+                            null
+
+                        override fun onOutboxDiscovered(
+                            event: AdvertisedRelayListEvent,
+                            relay: NormalizedRelayUrl,
+                        ) {
+                            buffered.add(event)
+                        }
+
+                        override fun onDiscoveredEvent(
+                            event: Event,
+                            relay: NormalizedRelayUrl,
+                        ) {
+                            buffered.add(event)
+                        }
+                    }
+
+                val dispatcher =
+                    OutboxDispatcher(
+                        client = ctx.client,
+                        scope = scope,
+                        indexRelays = { relays },
+                        gateway = gateway,
+                        overallTimeoutMs = overallTimeoutMs,
                     )
-                }
-            val received = ctx.drain(relays.associateWith { filters }, timeoutMs)
-            val kind3Events = received.mapNotNull { it.second as? ContactListEvent }
-            // Persist to store so future `get` / `list` see them.
-            kind3Events.forEach { runCatching { ctx.store.insert(it) } }
-            Output.emit(mapOf("received" to kind3Events.size, "followers" to follows.size))
-            return 0
+
+                val result = dispatcher.fetchKind3Only(follows)
+
+                // Persist to store so future `get` / `list` see them.
+                val eventsToPersist = synchronized(buffered) { buffered.toList() }
+                eventsToPersist.forEach { runCatching { ctx.store.insert(it) } }
+
+                Output.emit(
+                    mapOf(
+                        "followers" to follows.size,
+                        "authors_requested" to result.authorsRequested,
+                        "kind10002_received" to result.kind10002Received,
+                        "kind3_received" to result.kind3Received,
+                        "outbox_covered_authors" to result.outboxCoveredAuthors,
+                        "fallback_authors" to result.fallbackAuthors,
+                        "persisted" to eventsToPersist.size,
+                    ),
+                )
+                return 0
+            } finally {
+                scope.cancel()
+            }
         }
     }
 

desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/Main.kt

@@ -1576,16 +1576,32 @@ fun MainContent(
                 iAccount.wotService.applyKind3(evt.pubKey, evt.verifiedFollowKeySet())
             }
         }
-        // React to changes in the active user's follow set.
+        // React to changes in the active user's follow set. Under the
+        // outbox model (PR #3483 review directive from Vitor) kind-3
+        // fetch goes to each author's declared write relays instead of a
+        // static index-relay broadcast — the OutboxDispatcher does the
+        // NIP-65 discovery, transposes with RelayListRecommendationProcessor
+        // and issues per-outbox-relay REQs. Falls back to index relays
+        // for authors that never returned a 10002.
         launch {
             localCache.followedUsers.collect { follows ->
                 iAccount.wotService.onFollowSetChange(follows, account.pubKeyHex)
-                if (follows.isNotEmpty()) {
-                    subscriptionsCoordinator.loadKind3Batched(follows) {
+                when {
+                    iAccount.wotService.isDisabled.value -> {
+                        // Guardrail — mega-follow accounts skip WoT
+                        // entirely so we don't dispatch a batch that
+                        // would be discarded anyway.
                         iAccount.wotService.markReadyOnce()
                     }
-                } else {
-                    iAccount.wotService.markReadyOnce()
+                    follows.isEmpty() -> {
+                        iAccount.wotService.markReadyOnce()
+                    }
+                    else -> {
+                        launch {
+                            subscriptionsCoordinator.loadKind3ViaOutbox(follows)
+                            iAccount.wotService.markReadyOnce()
+                        }
+                    }
                 }
             }
         }

desktopApp/src/jvmMain/kotlin/com/vitorpamplona/amethyst/desktop/subscriptions/DesktopRelaySubscriptionsCoordinator.kt

@@ -25,6 +25,8 @@ import com.vitorpamplona.amethyst.commons.relayClient.assemblers.FeedMetadataCoo
 import com.vitorpamplona.amethyst.commons.relayClient.preload.MetadataPreloader
 import com.vitorpamplona.amethyst.commons.relayClient.preload.MetadataRateLimiter
 import com.vitorpamplona.amethyst.commons.service.BasicBundledInsert
+import com.vitorpamplona.amethyst.commons.wot.OutboxCacheGateway
+import com.vitorpamplona.amethyst.commons.wot.OutboxDispatcher
 import com.vitorpamplona.amethyst.desktop.cache.DesktopLocalCache
 import com.vitorpamplona.amethyst.desktop.model.DesktopDmRelayState
 import com.vitorpamplona.quartz.nip01Core.core.Event
@@ -33,6 +35,7 @@ import com.vitorpamplona.quartz.nip01Core.relay.client.INostrClient
 import com.vitorpamplona.quartz.nip01Core.relay.client.reqs.SubscriptionListener
 import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
 import com.vitorpamplona.quartz.nip01Core.relay.normalizer.NormalizedRelayUrl
+import com.vitorpamplona.quartz.nip65RelayList.AdvertisedRelayListEvent
 import com.vitorpamplona.quartz.utils.Log
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
@@ -98,6 +101,51 @@ class DesktopRelaySubscriptionsCoordinator(
             },
         )
 
+    /**
+     * Bridges [OutboxDispatcher] to [DesktopLocalCache]. Every event the
+     * dispatcher receives goes through [DesktopLocalCache.consume] so it
+     * lands in the same code path as events arriving from feed
+     * subscriptions — kind-10002 caches into `addressableNotes`; kind-0
+     * updates the user metadata; kind-3 fans out through
+     * `_contactListEvents` for the WoT service.
+     */
+    private val outboxGateway =
+        object : OutboxCacheGateway {
+            override fun cachedOutbox(pubkey: HexKey): AdvertisedRelayListEvent? = localCache.cachedAdvertisedRelayList(pubkey)
+
+            override fun onOutboxDiscovered(
+                event: AdvertisedRelayListEvent,
+                relay: NormalizedRelayUrl,
+            ) {
+                localCache.consume(event, relay)
+            }
+
+            override fun onDiscoveredEvent(
+                event: Event,
+                relay: NormalizedRelayUrl,
+            ) {
+                localCache.consume(event, relay)
+            }
+        }
+
+    /**
+     * NIP-65 outbox model for kind-0 and kind-3 fetching. Per PR #3483
+     * review directive from Vitor: index relays discover each author's
+     * write-relay list, then kind-0/kind-3 REQs go to that author's
+     * declared write relays. See [OutboxDispatcher] for the pipeline.
+     *
+     * Kept as a val (not lazy) because [clear] must reset its dedup
+     * markers on account switch. The dispatcher itself is stateless
+     * across accounts as long as `clear()` is called.
+     */
+    val outboxDispatcher =
+        OutboxDispatcher(
+            client = client,
+            scope = scope,
+            indexRelays = { indexRelays },
+            gateway = outboxGateway,
+        )
+
     // Event bundler: batches consumed notes before emitting to SharedFlow
     // 250ms for desktop (Android uses 1000ms to save battery)
     private val eventBundler =
@@ -387,10 +435,20 @@ class DesktopRelaySubscriptionsCoordinator(
 
         unsubscribeFromDms()
         feedMetadata.clear()
+        outboxDispatcher.clear()
         rateLimiter.reset()
         cleanupJob?.cancel()
     }
 
+    /**
+     * Fetch kind-3 (follow lists) for [pubkeys] via each author's outbox
+     * relay per NIP-65 (see [OutboxDispatcher]). Suspends until every
+     * phase EOSEs or times out. Callers typically launch this on a
+     * scope-owned coroutine and mark the WoT service ready in the
+     * continuation. Returns per-phase counters for observability.
+     */
+    suspend fun loadKind3ViaOutbox(pubkeys: Set<HexKey>): OutboxDispatcher.Result = outboxDispatcher.fetchKind3Only(pubkeys)
+
     // ----- Memory Cleanup -----
 
     private val memoryBean = ManagementFactory.getMemoryMXBean()