Say why delivery failed, instead of only logging it (ios#21, #22, #23)

The Android half, matching the iOS change.

**The give-up rule (ios#22).** `flushOutbox` used
`runCatching { … }.getOrNull()`, which threw the exception away — so the loop
could not tell a contact who is merely offline from one whose identity keys
changed, and retried both for ever while the bubble went on promising
"will send". The failure is caught now and classified: a refusal from the
trust layer stops, marks the bubble FAILED and records why. Everything else
still retries for ever, deliberately — a contact who is offline or has no
device yet gets the message whenever they return.

⚠️ Classified on the exception TYPE, never on message text. A string match
would break the first time somebody reworded an error, and would break
silently: the queue would simply go back to retrying for ever.

**The reading (ios#21, #23).** `DeliveryDiagnostics` counts envelopes seen,
drops by reason, session repairs and the last send failure, per conversation.
Three or more repairs against one peer is flagged — a client that repeatedly
repairs is a symptom, not a cure.

🔴 **Shape only.** Refusal messages name what was wrong, which sometimes
means naming a device id: "device ce0c0fb4 has published keys, but none
match". Those are normalised before being counted, so the reading is safe to
screenshot and the map cannot grow a row per device id. ⚠️ Short hex ids like
`9f0e` slip past the obvious rules; the test covers them, and covers that
"base64" survives unmangled.

The new outbox test was checked by disabling the give-up rule: it fails
without it.

581 tests pass.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

app/src/main/kotlin/io/flamenet/messenger/app/Session.kt

@@ -1033,6 +1033,32 @@ class Session(
      * order to tidy up a label, and the label is not the point — the message
      * is. A contact who reinstalls in a month gets it in a month.
      */
+    /**
+     * Whether waiting will ever resolve this, mirroring iOS `E2EError.waitingCanFix`.
+     *
+     * ⚠️ Matched on the refusal types the trust layer throws, not on message
+     * text. A string match here would break the first time somebody reworded
+     * an error, and would break silently — the queue would simply go back to
+     * retrying for ever.
+     */
+    private fun permanentRefusal(e: Throwable?): Boolean = when (e) {
+        is io.flamenet.client.IdentityChangedException,
+        is io.flamenet.client.KeyNotLoggedException,
+        is io.flamenet.client.KeyLogForkException -> true
+        else -> false
+    }
+
+    /** A short, contentless reason for the reading and the bubble. */
+    private fun refusalReason(e: Throwable?): String = when (e) {
+        is io.flamenet.client.IdentityChangedException ->
+            "their security keys changed"
+        is io.flamenet.client.KeyNotLoggedException ->
+            "their device offered a key the server has no record of"
+        is io.flamenet.client.KeyLogForkException ->
+            "this server's record of published keys does not add up"
+        else -> "the message could not be sent"
+    }
+
     fun flushOutbox(): Set<String> {
         val pending = vault.outbox()
         if (pending.length() == 0) return emptySet()
@@ -1059,7 +1085,12 @@ class Session(
                 continue
             }
 
-            val sent = runCatching { conversations.send(to, text) }.getOrNull()
+            // ⚠️ The failure is CAUGHT, not discarded. `getOrNull()` threw the
+            // exception away, so this loop could not tell a contact who is
+            // merely offline from one whose keys changed — and retried both
+            // for ever.
+            val attempt = runCatching { conversations.send(to, text) }
+            val sent = attempt.getOrNull()
             if (sent?.reachedSomeone == true) {
                 if (groupThread == null) {
                     restate(Vault.threadForUser(to), at, Delivery.DELIVERED)
@@ -1067,6 +1098,21 @@ class Session(
                 } else {
                     groupDelivered += groupThread to at
                 }
+            } else if (permanentRefusal(attempt.exceptionOrNull())) {
+                // 🔴 **Some refusals waiting cannot fix** (ios#22). A send
+                // blocked because the contact's identity keys changed is
+                // refused identically on every retry until a person accepts
+                // them, while the bubble goes on promising "will send".
+                //
+                // ⚠️ Everything else still retries for ever, deliberately: a
+                // contact who is offline, or who has not registered a device
+                // yet, gets the message whenever they return. Giving up there
+                // would delete something somebody wrote to tidy up a label.
+                val why = refusalReason(attempt.exceptionOrNull())
+                io.flamenet.client.DeliveryDiagnostics.sendFailed(why, to)
+                val thread = groupThread ?: Vault.threadForUser(to)
+                restate(thread, at, Delivery.FAILED)
+                touched += thread
             } else {
                 stillWaiting.put(item)
             }

app/src/test/kotlin/io/flamenet/messenger/app/OutboxTest.kt

@@ -153,6 +153,67 @@ class OutboxTest {
                      "a transport failure must not eat the message either")
     }
 
+    /**
+     * 🔴 **A refusal waiting cannot fix must STOP** (ios#22).
+     *
+     * The outbox retries for ever, which is right for a contact who is offline
+     * or has no device yet. It is wrong for a send the trust layer refuses:
+     * the peer's identity keys changed, so every retry is refused identically
+     * until a person accepts them — while the bubble goes on promising
+     * "will send", which is a promise the app cannot keep.
+     *
+     * ⚠️ Asserted on the OUTBOX and the BUBBLE together. A version that
+     * dropped the entry but left the bubble pending would pass a test that
+     * only checked one, and would read as "still sending" for ever.
+     */
+    @Test
+    fun `a send refused for changed keys stops retrying and says why`() {
+        val session = session()
+        server.enqueue(json("""{"devices":[]}"""))
+        session.send(peer, "you there?")
+        assertEquals(1, session.vault.outbox().length())
+
+        // They come back with a REAL device, and the message goes — pinning
+        // their identity on first use.
+        val them = DeviceIdentity.create()
+        val bundleOf = { id: DeviceIdentity ->
+            val opk = id.generateOneTimePreKeys(1).first()
+            id.registrationWire().put(
+                "opk", org.json.JSONObject()
+                    .put("id", opk.id)
+                    .put("pub", android.util.Base64.encodeToString(
+                        opk.publicKey, android.util.Base64.NO_WRAP)))
+        }
+        val first = bundleOf(them)
+        server.enqueue(json(org.json.JSONObject()
+            .put("devices", org.json.JSONArray().put(first)).toString()))
+        server.enqueue(json(org.json.JSONObject()
+            .put("devices", org.json.JSONArray().put(first)).toString()))
+        server.enqueue(json("""{"ok":true}"""))
+        session.flushOutbox()
+        assertEquals(0, session.vault.outbox().length())
+
+        // 🔴 Now they reinstall: a DIFFERENT identity under the same user. The
+        // trust layer refuses, and no amount of retrying changes that.
+        server.enqueue(json("""{"devices":[]}"""))      // queues the second message
+        session.send(peer, "you there?")
+        assertEquals(1, session.vault.outbox().length())
+
+        val second = bundleOf(DeviceIdentity.create())
+        val newDirectory = org.json.JSONObject()
+            .put("devices", org.json.JSONArray().put(second)).toString()
+        // Enqueued generously: the refusal happens at the directory, but the
+        // exact number of calls before it is not what this test is about.
+        repeat(4) { server.enqueue(json(newDirectory)) }
+        session.flushOutbox()
+
+        assertEquals(0, session.vault.outbox().length(),
+                     "a permanently refused send must not sit in the queue for ever")
+        val line = session.history(Vault.threadForUser(peer)).last { it.mine }
+        assertEquals(Session.Delivery.FAILED, line.delivery,
+                     "the bubble must stop promising it will send")
+    }
+
     /**
      * 🔴 **The promise, end to end.** Everything above only proves the message
      * is not lost; this proves it is eventually sent, which is the part the