Merge remote-tracking branch 'origin/main' into codex/nostr-double-ratchet

AGENTS.md

@@ -56,6 +56,11 @@ The application follows a clean architecture pattern, heavily modularized by fea
 ### Testing
 - **Unit Tests**: Located in `app/src/test/`. Use for business logic, protocols, and utility testing.
 - **Instrumented Tests**: Located in `app/src/androidTest/`. Use for UI and permission integration testing.
+- **Device Mesh Tests (ADB test hooks)**: Two-physical-device scenarios driven over ADB, **kept separate from Gradle/CI** — run them manually when changing mesh/crypto/transfer code. A debug-only broadcast receiver (`app/src/debug/java/com/bitchat/android/testhook/`, never in release builds) exposes mesh operations (scan, connect, Noise handshake, DMs, broadcast, files, raw packet injection) via `am broadcast -a com.bitchat.droid.TEST_HOOK`; the host orchestrator is `tools/release_gate/mesh_lab.py`. Full guide: `docs/release-gate-runbook.md` appendix "mesh lab".
+  - Prereqs: `adb` on PATH, Python 3.10+, two devices with USB debugging, **both unlocked with screen on** (locked/dozing → POWER_SAVER → flaky timing).
+  - Setup: `./gradlew assembleDebug && python3 tools/release_gate/mesh_lab.py setup --serial-a <s1> --serial-b <s2> --apk app/build/outputs/apk/debug/app-arm64-v8a-debug.apk`
+  - Run: `python3 tools/release_gate/mesh_lab.py scenario all --serial-a <s1> --serial-b <s2> --out /tmp/meshlab-evidence`
+  - Scenarios: `dm`, `broadcast`, `file`, `file_oversize`, `file_private`, `raw`, `session_recovery`, `identity_reset`, `all`. Ad-hoc: `... cmd --serial <s> state`.
 - **Execution**:
   - Unit: `./gradlew test`
   - Instrumented: `./gradlew connectedAndroidTest`

app/src/debug/AndroidManifest.xml

@@ -0,0 +1,19 @@
+<?xml version="1.0" encoding="utf-8"?>
+<manifest xmlns:android="http://schemas.android.com/apk/res/android"
+    xmlns:tools="http://schemas.android.com/tools">
+
+    <application>
+        <!-- Debug-only ADB test hook. Drives mesh operations (scan, connect,
+             handshake, DMs, files, broadcast, raw packets) from host scripts.
+             Exported intentionally so `adb shell am broadcast` can reach it;
+             never shipped in release builds. -->
+        <receiver
+            android:name="com.bitchat.android.testhook.TestHookReceiver"
+            android:exported="true"
+            tools:ignore="ExportedReceiver">
+            <intent-filter>
+                <action android:name="com.bitchat.droid.TEST_HOOK" />
+            </intent-filter>
+        </receiver>
+    </application>
+</manifest>

app/src/debug/java/com/bitchat/android/testhook/TestHookDriver.kt

@@ -0,0 +1,495 @@
+package com.bitchat.android.testhook
+
+import android.content.Context
+import android.content.Intent
+import android.util.Log
+import com.bitchat.android.features.file.FileUtils
+import com.bitchat.android.mesh.MeshService
+import com.bitchat.android.mesh.PrivateMediaPreparation
+import com.bitchat.android.mesh.TransferProgressManager
+import com.bitchat.android.model.BitchatFilePacket
+import com.bitchat.android.model.RoutedPacket
+import com.bitchat.android.noise.NoiseSession
+import com.bitchat.android.protocol.BitchatPacket
+import com.bitchat.android.service.MeshForegroundService
+import com.bitchat.android.service.MeshServiceHolder
+import com.bitchat.android.service.TransportBridgeService
+import com.bitchat.android.services.AppStateStore
+import com.bitchat.android.ui.DataManager
+import com.bitchat.android.util.AppConstants
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.async
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.withContext
+import kotlinx.coroutines.withTimeout
+import kotlinx.coroutines.withTimeoutOrNull
+import org.json.JSONArray
+import org.json.JSONObject
+import java.io.File
+import java.security.MessageDigest
+
+/**
+ * Headless engine behind [TestHookReceiver]. Drives the public [MeshService] API and
+ * observes state via [AppStateStore] flows (never touches the single-slot mesh delegate).
+ */
+object TestHookDriver {
+
+    private const val TAG = TestHookReceiver.TAG
+
+    private const val DEFAULT_SCAN_TIMEOUT_MS = 30_000L
+    private const val DEFAULT_CONNECT_TIMEOUT_MS = 30_000L
+    private const val DEFAULT_HANDSHAKE_TIMEOUT_MS = 30_000L
+    private const val DEFAULT_RECV_TIMEOUT_MS = 60_000L
+    private const val DEFAULT_FILE_TIMEOUT_MS = 180_000L
+
+    suspend fun execute(context: Context, cmd: String, intent: Intent): JSONObject {
+        Log.d(TAG, "execute cmd=$cmd")
+        val result = when (cmd) {
+            "ping" -> ok(cmd).put("pong", true).put("package", context.packageName)
+            "start" -> start(context)
+            "stop" -> stop(context)
+            "whoami" -> whoami(context)
+            "set_nickname" -> setNickname(context, intent.requiredString("name"))
+            "scan" -> scan(context, intent)
+            "peers" -> peers(context)
+            "connect" -> connect(intent.requiredString("peer"), intent)
+            "handshake" -> handshake(context, intent.requiredString("peer"), intent)
+            "session" -> session(context, intent.requiredString("peer"))
+            "announce" -> announce(context)
+            "broadcast_msg" -> broadcastMsg(context, intent.requiredString("content"), intent.getStringExtra("channel"))
+            "dm_send" -> dmSend(context, intent.requiredString("peer"), intent.requiredString("content"), intent.getStringExtra("msg_id"))
+            "dm_recv" -> dmRecv(context, intent)
+            "msg_recv" -> msgRecv(context, intent)
+            "file_send" -> fileSend(context, intent)
+            "file_recv" -> fileRecv(context, intent)
+            "file_cancel" -> fileCancel(context, intent.requiredString("transfer_id"))
+            "raw_send" -> rawSend(context, intent)
+            "ble" -> setBle(intent.getBooleanExtra("enabled", true))
+            "state" -> state(context)
+            "clear_results" -> clearResults(context)
+            else -> err(cmd, "unknown command: $cmd")
+        }
+        return result.put("cmd", cmd)
+    }
+
+    // MARK: - Lifecycle
+
+    private fun start(context: Context): JSONObject {
+        MeshForegroundService.start(context)
+        val mesh = mesh(context)
+        mesh.startServices()
+        return ok("start").put("peer_id", mesh.myPeerID)
+    }
+
+    private fun stop(context: Context): JSONObject {
+        try {
+            MeshServiceHolder.unifiedMeshService?.stopServices()
+        } catch (e: Exception) {
+            Log.w(TAG, "stopServices failed: ${e.message}")
+        }
+        MeshForegroundService.stop(context)
+        return ok("stop")
+    }
+
+    // MARK: - Identity
+
+    private fun whoami(context: Context): JSONObject {
+        val mesh = mesh(context)
+        return ok("whoami")
+            .put("peer_id", mesh.myPeerID)
+            .put("identity_fingerprint", mesh.getIdentityFingerprint())
+            .put("noise_public_key", mesh.getStaticNoisePublicKey()?.toHex())
+            .put("nickname", AppStateStore.nickname.value)
+    }
+
+    private fun setNickname(context: Context, name: String): JSONObject {
+        DataManager(context).saveNickname(name)
+        AppStateStore.setNickname(name)
+        mesh(context).sendBroadcastAnnounce()
+        return ok("set_nickname").put("nickname", name)
+    }
+
+    // MARK: - Discovery / connection
+
+    private suspend fun scan(context: Context, intent: Intent): JSONObject {
+        val timeoutMs = intent.getLongExtra("timeout_ms", DEFAULT_SCAN_TIMEOUT_MS)
+        val minPeers = intent.getIntExtra("min_peers", 1)
+        val mesh = mesh(context)
+        val found = withTimeoutOrNull(timeoutMs) {
+            AppStateStore.peers.first { it.size >= minPeers }
+        }
+        val peerIds = found ?: AppStateStore.peers.value
+        return ok("scan")
+            .put("reached_min_peers", found != null)
+            .put("peers", peerInfosJson(mesh, peerIds))
+    }
+
+    private fun peers(context: Context): JSONObject {
+        val mesh = mesh(context)
+        return ok("peers").put("peers", peerInfosJson(mesh, AppStateStore.peers.value))
+    }
+
+    private suspend fun connect(peerID: String, intent: Intent): JSONObject {
+        val timeoutMs = intent.getLongExtra("timeout_ms", DEFAULT_CONNECT_TIMEOUT_MS)
+        val ble = MeshServiceHolder.meshService ?: return err("connect", "BLE service not running")
+        val address = ble.getDeviceAddressForPeer(peerID)
+            ?: return err("connect", "no device address known for peer $peerID (scan first)")
+        val accepted = ble.connectionManager.connectToAddress(address)
+        if (!accepted) return err("connect", "connectToAddress($address) rejected")
+        val direct = withTimeoutOrNull(timeoutMs) {
+            AppStateStore.directPeers.first { it.contains(peerID) }
+        }
+        return ok("connect")
+            .put("peer", peerID)
+            .put("address", address)
+            .put("direct", direct != null)
+    }
+
+    // MARK: - Noise
+
+    private suspend fun handshake(context: Context, peerID: String, intent: Intent): JSONObject {
+        val timeoutMs = intent.getLongExtra("timeout_ms", DEFAULT_HANDSHAKE_TIMEOUT_MS)
+        val mesh = mesh(context)
+        val deadline = System.currentTimeMillis() + timeoutMs
+        if (!mesh.hasEstablishedSession(peerID)) {
+            mesh.initiateNoiseHandshake(peerID)
+        }
+        var lastState: NoiseSession.NoiseSessionState = NoiseSession.NoiseSessionState.Uninitialized
+        while (System.currentTimeMillis() < deadline) {
+            lastState = mesh.getSessionState(peerID)
+            when (lastState) {
+                is NoiseSession.NoiseSessionState.Established -> {
+                    return ok("handshake")
+                        .put("peer", peerID)
+                        .put("state", lastState.toString())
+                        .put("fingerprint", mesh.getPeerFingerprint(peerID))
+                }
+                is NoiseSession.NoiseSessionState.Failed -> {
+                    return err("handshake", "session failed: $lastState").put("peer", peerID)
+                }
+                else -> delay(100)
+            }
+        }
+        return err("handshake", "timeout after ${timeoutMs}ms (last state: $lastState)").put("peer", peerID)
+    }
+
+    private fun session(context: Context, peerID: String): JSONObject {
+        val mesh = mesh(context)
+        return ok("session")
+            .put("peer", peerID)
+            .put("state", mesh.getSessionState(peerID).toString())
+            .put("established", mesh.hasEstablishedSession(peerID))
+            .put("fingerprint", mesh.getPeerFingerprint(peerID))
+    }
+
+    // MARK: - Messaging
+
+    private fun announce(context: Context): JSONObject {
+        mesh(context).sendBroadcastAnnounce()
+        return ok("announce")
+    }
+
+    private fun broadcastMsg(context: Context, content: String, channel: String?): JSONObject {
+        mesh(context).sendMessage(content, emptyList(), channel)
+        return ok("broadcast_msg").put("content", content).put("channel", channel)
+    }
+
+    private fun dmSend(context: Context, peerID: String, content: String, msgID: String?): JSONObject {
+        val mesh = mesh(context)
+        val nickname = mesh.getPeerNicknames()[peerID] ?: peerID
+        val id = msgID ?: "testhook-${System.currentTimeMillis()}"
+        mesh.sendPrivateMessage(content, peerID, nickname, id)
+        return ok("dm_send").put("peer", peerID).put("msg_id", id)
+    }
+
+    private suspend fun dmRecv(context: Context, intent: Intent): JSONObject {
+        val timeoutMs = intent.getLongExtra("timeout_ms", DEFAULT_RECV_TIMEOUT_MS)
+        val fromPeer = intent.getStringExtra("peer")
+        val contains = intent.getStringExtra("contains")
+        val startTime = System.currentTimeMillis()
+        val mesh = mesh(context)
+        val match = withTimeoutOrNull(timeoutMs) {
+            AppStateStore.privateMessages.first { conversations ->
+                conversations.values.flatten().any { msg ->
+                    msg.timestamp.time >= startTime &&
+                        msg.senderPeerID != mesh.myPeerID &&
+                        (fromPeer == null || msg.senderPeerID == fromPeer) &&
+                        (contains == null || msg.content.contains(contains))
+                }
+            }
+        } ?: return err("dm_recv", "timeout after ${timeoutMs}ms")
+        val msg = match.values.flatten().first { msg ->
+            msg.timestamp.time >= startTime &&
+                msg.senderPeerID != mesh.myPeerID &&
+                (fromPeer == null || msg.senderPeerID == fromPeer) &&
+                (contains == null || msg.content.contains(contains))
+        }
+        return ok("dm_recv")
+            .put("from", msg.senderPeerID)
+            .put("sender", msg.sender)
+            .put("content", msg.content)
+            .put("msg_id", msg.id)
+    }
+
+    private suspend fun msgRecv(context: Context, intent: Intent): JSONObject {
+        val timeoutMs = intent.getLongExtra("timeout_ms", DEFAULT_RECV_TIMEOUT_MS)
+        val contains = intent.getStringExtra("contains")
+        val channel = intent.getStringExtra("channel")
+        val startTime = System.currentTimeMillis()
+        val mesh = mesh(context)
+        val matches: (com.bitchat.android.model.BitchatMessage) -> Boolean = { msg ->
+            msg.timestamp.time >= startTime &&
+                msg.senderPeerID != mesh.myPeerID &&
+                (contains == null || msg.content.contains(contains)) &&
+                (channel == null || msg.channel == channel)
+        }
+        val found = withTimeoutOrNull(timeoutMs) {
+            if (channel != null) {
+                AppStateStore.channelMessages.first { m -> m.values.flatten().any(matches) }
+                    .values.flatten().first(matches)
+            } else {
+                AppStateStore.publicMessages.first { l -> l.any(matches) }.first(matches)
+            }
+        } ?: return err("msg_recv", "timeout after ${timeoutMs}ms")
+        return ok("msg_recv")
+            .put("from", found.senderPeerID)
+            .put("sender", found.sender)
+            .put("content", found.content)
+            .put("channel", found.channel)
+            .put("msg_id", found.id)
+    }
+
+    // MARK: - File transfer
+
+    private suspend fun fileSend(context: Context, intent: Intent): JSONObject {
+        val path = intent.requiredString("path")
+        val peerID = intent.getStringExtra("peer")
+        val timeoutMs = intent.getLongExtra("timeout_ms", DEFAULT_FILE_TIMEOUT_MS)
+        val mesh = mesh(context)
+
+        val file = File(path)
+        if (!file.isFile) return err("file_send", "file not found: $path")
+        val content = withContext(Dispatchers.IO) { file.readBytes() }
+        if (content.size.toLong() > AppConstants.Media.MAX_FILE_SIZE_BYTES) {
+            return err("file_send", "file too large: ${content.size} > ${AppConstants.Media.MAX_FILE_SIZE_BYTES}")
+        }
+        val packet = BitchatFilePacket(
+            fileName = file.name,
+            fileSize = content.size.toLong(),
+            mimeType = intent.getStringExtra("mime") ?: FileUtils.getMimeTypeFromExtension(file.name),
+            content = content
+        )
+        val encoded = packet.encode() ?: return err("file_send", "failed to TLV-encode packet")
+        val transferId = sha256Hex(encoded)
+
+        return coroutineScope {
+            // Subscribe on a background dispatcher before sending so synchronous
+            // failure events are not missed (SharedFlow has replay=0).
+            val completion = async(Dispatchers.Default) {
+                TransferProgressManager.events.first { it.transferId == transferId && it.completed }
+            }
+            delay(50)
+            val sendError = dispatchFileSend(context, intent, mesh, peerID, packet, transferId)
+            if (sendError != null) {
+                completion.cancel()
+                return@coroutineScope sendError.put("cmd", "file_send")
+            }
+            val event = withTimeoutOrNull(timeoutMs) { completion.await() }
+                ?: return@coroutineScope err("file_send", "timeout waiting for transfer completion ($transferId)")
+            if (event.failed) {
+                return@coroutineScope err("file_send", "transfer rejected/failed before send ($transferId)")
+                    .put("transfer_id", transferId)
+            }
+            ok("file_send")
+                .put("transfer_id", transferId)
+                .put("sent", event.sent)
+                .put("total", event.total)
+                .put("bytes", content.size)
+                .put("peer", peerID)
+        }
+    }
+
+    private suspend fun dispatchFileSend(
+        context: Context,
+        intent: Intent,
+        mesh: MeshService,
+        peerID: String?,
+        packet: BitchatFilePacket,
+        transferId: String
+    ): JSONObject? {
+        if (peerID == null) {
+            mesh.sendFileBroadcast(packet)
+            return null
+        }
+        if (!mesh.hasEstablishedSession(peerID)) {
+            val hs = handshake(context, peerID, intent)
+            if (hs.optString("status") != "ok") return hs
+        }
+        // Peer state (capabilities/identity) can lag session establishment;
+        // retry transient preparation states before giving up.
+        val prepDeadline = System.currentTimeMillis() + 30_000
+        while (true) {
+            when (val prep = mesh.prepareFilePrivate(peerID, packet, transferId, allowLegacyFallback = false)) {
+                is PrivateMediaPreparation.Ready -> {
+                    return if (prep.transfer.commit()) null else err("file_send", "private transfer commit failed")
+                }
+                PrivateMediaPreparation.AwaitingPeerState,
+                PrivateMediaPreparation.NeedsHandshake -> {
+                    if (System.currentTimeMillis() >= prepDeadline) {
+                        return err("file_send", "private media preparation stuck at: $prep")
+                    }
+                    if (prep == PrivateMediaPreparation.NeedsHandshake) {
+                        mesh.initiateNoiseHandshake(peerID)
+                    }
+                    delay(500)
+                }
+                else -> return err("file_send", "private media preparation: $prep")
+            }
+        }
+    }
+
+    private suspend fun fileRecv(context: Context, intent: Intent): JSONObject {
+        val timeoutMs = intent.getLongExtra("timeout_ms", DEFAULT_FILE_TIMEOUT_MS)
+        val nameContains = intent.getStringExtra("name_contains")
+        val startTime = System.currentTimeMillis()
+        val dirs = listOf(
+            File(context.cacheDir, "files/incoming"),
+            File(context.cacheDir, "images/incoming")
+        )
+        val deadline = System.currentTimeMillis() + timeoutMs
+        while (System.currentTimeMillis() < deadline) {
+            val candidate = dirs
+                .flatMap { it.listFiles()?.toList() ?: emptyList() }
+                .filter { it.lastModified() >= startTime - 5_000 }
+                .filter { nameContains == null || it.name.contains(nameContains) }
+                .maxByOrNull { it.lastModified() }
+            if (candidate != null) {
+                val size1 = candidate.length()
+                delay(500)
+                if (candidate.length() == size1 && size1 > 0) {
+                    return ok("file_recv")
+                        .put("path", candidate.absolutePath)
+                        .put("name", candidate.name)
+                        .put("bytes", size1)
+                        .put("sha256", withContext(Dispatchers.IO) { sha256Hex(candidate.readBytes()) })
+                }
+            }
+            delay(250)
+        }
+        return err("file_recv", "timeout after ${timeoutMs}ms")
+    }
+
+    private fun fileCancel(context: Context, transferId: String): JSONObject {
+        val cancelled = mesh(context).cancelFileTransfer(transferId)
+        return ok("file_cancel").put("transfer_id", transferId).put("cancelled", cancelled)
+    }
+
+    // MARK: - Raw packet injection
+
+    private fun rawSend(context: Context, intent: Intent): JSONObject {
+        val payloadHex = intent.requiredString("payload_hex")
+        val typeStr = intent.requiredString("type")
+        val peerID = intent.getStringExtra("peer")
+        val ttl = intent.getIntExtra("ttl", 7)
+        val type = typeStr.toUIntOrNull(16)?.toUByte()
+            ?: return err("raw_send", "invalid type hex: $typeStr")
+        val payload = hexToBytes(payloadHex)
+            ?: return err("raw_send", "invalid payload_hex")
+        val mesh = mesh(context)
+        val packet = BitchatPacket(
+            type = type,
+            ttl = ttl.toUByte(),
+            senderID = mesh.myPeerID,
+            payload = payload
+        )
+        if (peerID != null) {
+            TransportBridgeService.sendToPeerFromLocal(peerID, packet)
+        } else {
+            TransportBridgeService.broadcastFromLocal(RoutedPacket(packet))
+        }
+        return ok("raw_send")
+            .put("type", typeStr)
+            .put("payload_bytes", payload.size)
+            .put("peer", peerID)
+    }
+
+    // MARK: - Transport / state
+
+    private fun setBle(enabled: Boolean): JSONObject {
+        val ble = MeshServiceHolder.meshService ?: return err("ble", "BLE service not running")
+        ble.setBleTransportEnabled(enabled)
+        return ok("ble").put("enabled", enabled)
+    }
+
+    private fun state(context: Context): JSONObject {
+        val mesh = mesh(context)
+        val peersJson = peerInfosJson(mesh, AppStateStore.peers.value)
+        val sessions = JSONObject()
+        AppStateStore.peers.value.forEach { peerID ->
+            sessions.put(peerID, mesh.getSessionState(peerID).toString())
+        }
+        return ok("state")
+            .put("peer_id", mesh.myPeerID)
+            .put("nickname", AppStateStore.nickname.value)
+            .put("peers", peersJson)
+            .put("direct_peers", JSONArray(AppStateStore.directPeers.value.toList()))
+            .put("sessions", sessions)
+            .put("device_map", JSONObject(mesh.getDeviceAddressToPeerMapping() as Map<*, *>))
+            .put("debug_status", mesh.getDebugStatus())
+    }
+
+    private fun clearResults(context: Context): JSONObject {
+        val dir = File(context.cacheDir, "testhook/results")
+        val count = dir.listFiles()?.count { it.delete() } ?: 0
+        return ok("clear_results").put("deleted", count)
+    }
+
+    // MARK: - Helpers
+
+    private fun mesh(context: Context): MeshService = MeshServiceHolder.getUnifiedOrCreate(context)
+
+    private fun peerInfosJson(mesh: MeshService, peerIds: List<String>): JSONArray {
+        val nicknames = mesh.getPeerNicknames()
+        val rssi = mesh.getPeerRSSI()
+        val arr = JSONArray()
+        peerIds.forEach { id ->
+            val info = mesh.getPeerInfo(id)
+            arr.put(JSONObject()
+                .put("id", id)
+                .put("nickname", nicknames[id] ?: info?.nickname)
+                .put("rssi", rssi[id])
+                .put("direct", AppStateStore.directPeers.value.contains(id))
+                .put("connected", info?.isConnected)
+                .put("last_seen", info?.lastSeen)
+                .put("session", mesh.getSessionState(id).toString())
+                .put("fingerprint", mesh.getPeerFingerprint(id)))
+        }
+        return arr
+    }
+
+    private fun ok(cmd: String) = JSONObject().put("status", "ok").put("cmd", cmd)
+    private fun err(cmd: String, message: String) =
+        JSONObject().put("status", "error").put("cmd", cmd).put("error", message)
+
+    private fun Intent.requiredString(name: String): String =
+        getStringExtra(name) ?: throw IllegalArgumentException("missing required extra: $name")
+
+    private fun sha256Hex(data: ByteArray): String =
+        MessageDigest.getInstance("SHA-256").digest(data).toHex()
+
+    private fun ByteArray.toHex(): String = joinToString("") { "%02x".format(it) }
+
+    private fun hexToBytes(hex: String): ByteArray? {
+        val clean = hex.replace(" ", "")
+        if (clean.length % 2 != 0) return null
+        return try {
+            ByteArray(clean.length / 2) { i ->
+                clean.substring(i * 2, i * 2 + 2).toInt(16).toByte()
+            }
+        } catch (e: Exception) {
+            null
+        }
+    }
+}

app/src/debug/java/com/bitchat/android/testhook/TestHookReceiver.kt

@@ -0,0 +1,65 @@
+package com.bitchat.android.testhook
+
+import android.content.BroadcastReceiver
+import android.content.Context
+import android.content.Intent
+import android.util.Log
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import org.json.JSONObject
+import java.io.File
+
+/**
+ * ADB-drivable test hook (debug builds only).
+ *
+ * Usage:
+ *   adb shell am broadcast -a com.bitchat.droid.TEST_HOOK \
+ *     --es cmd <command> --es id <cmd-id> [command extras...]
+ *
+ * Result is written to cache/testhook/results/<id>.json and logged under tag TestHook:
+ *   adb shell run-as com.bitchat.droid cat cache/testhook/results/<id>.json
+ */
+class TestHookReceiver : BroadcastReceiver() {
+
+    companion object {
+        const val TAG = "TestHook"
+        const val ACTION = "com.bitchat.droid.TEST_HOOK"
+        private const val DEFAULT_OVERALL_TIMEOUT_MS = 180_000L
+    }
+
+    override fun onReceive(context: Context, intent: Intent) {
+        if (intent.action != ACTION) return
+        val cmd = intent.getStringExtra("cmd") ?: "ping"
+        val id = intent.getStringExtra("id") ?: "cmd-${System.currentTimeMillis()}"
+        val overallTimeout = intent.getLongExtra("overall_timeout_ms", DEFAULT_OVERALL_TIMEOUT_MS)
+
+        Log.i(TAG, "CMD id=$id cmd=$cmd")
+
+        val pendingResult = goAsync()
+        Thread {
+            val result = try {
+                runBlocking {
+                    withTimeout(overallTimeout) {
+                        TestHookDriver.execute(context.applicationContext, cmd, intent)
+                    }
+                }
+            } catch (e: Exception) {
+                JSONObject()
+                    .put("status", "error")
+                    .put("cmd", cmd)
+                    .put("error", "${e.javaClass.simpleName}: ${e.message}")
+            }
+            try {
+                val dir = File(context.cacheDir, "testhook/results").apply { mkdirs() }
+                File(dir, "$id.json").writeText(result.toString())
+            } catch (e: Exception) {
+                Log.e(TAG, "Failed to write result file for $id: ${e.message}")
+            }
+            Log.i(TAG, "RESULT id=$id $result")
+        }.start()
+        // Finish immediately: long-running commands continue on the worker thread and
+        // report via the result file. Holding the broadcast open past the system
+        // broadcast window would ANR the app.
+        pendingResult.finish()
+    }
+}

app/src/main/java/com/bitchat/android/mesh/FragmentingPacketSender.kt

@@ -32,7 +32,13 @@ class FragmentingPacketSender(
         sendSingle: (RoutedPacket) -> Boolean
     ): Boolean {
         val transferId = transferIdFor(routed)
-        val packets = packetsForTransport(routed) ?: return false
+        val packets = packetsForTransport(routed)
+        if (packets == null) {
+            if (transferId != null) {
+                TransferProgressManager.fail(transferId)
+            }
+            return false
+        }
         val total = packets.size
 
         if (total <= 1) {
@@ -46,9 +52,13 @@ class FragmentingPacketSender(
                     preparedPackets = null
                 )
             )
-            if (sent && transferId != null) {
-                TransferProgressManager.progress(transferId, 1, 1)
-                TransferProgressManager.complete(transferId, 1)
+            if (transferId != null) {
+                if (sent) {
+                    TransferProgressManager.progress(transferId, 1, 1)
+                    TransferProgressManager.complete(transferId, 1)
+                } else {
+                    TransferProgressManager.fail(transferId)
+                }
             }
             return sent
         }
@@ -197,7 +207,12 @@ class FragmentingPacketSender(
 
         val manager = fragmentManager ?: return listOf(packet)
         return try {
-            val fragments = manager.createFragments(packet)
+            // Receivers hard-cap reassembly at MAX_FRAGMENTS_PER_ID; sending more
+            // fragments would be undeliverable, so reject here instead.
+            val fragments = manager.createFragments(
+                packet,
+                com.bitchat.android.util.AppConstants.Fragmentation.MAX_FRAGMENTS_PER_ID
+            )
             if (fragments.isEmpty()) {
                 Log.e(logTag, "Fragment manager returned no packets for packet type ${packet.type}")
                 null

app/src/main/java/com/bitchat/android/mesh/TransferProgressManager.kt

@@ -11,7 +11,8 @@ data class TransferProgressEvent(
     val transferId: String,
     val sent: Int,
     val total: Int,
-    val completed: Boolean
+    val completed: Boolean,
+    val failed: Boolean = false
 )
 
 object TransferProgressManager {
@@ -22,9 +23,9 @@ object TransferProgressManager {
     fun start(id: String, total: Int) { emit(id, 0, total, false) }
     fun progress(id: String, sent: Int, total: Int) { emit(id, sent, total, sent >= total) }
     fun complete(id: String, total: Int) { emit(id, total, total, true) }
+    fun fail(id: String) { emit(id, 0, 0, done = true, failed = true) }
 
-    private fun emit(id: String, sent: Int, total: Int, done: Boolean) {
-        scope.launch { _events.emit(TransferProgressEvent(id, sent, total, done)) }
+    private fun emit(id: String, sent: Int, total: Int, done: Boolean, failed: Boolean = false) {
+        scope.launch { _events.emit(TransferProgressEvent(id, sent, total, done, failed)) }
     }
 }
-

app/src/main/java/com/bitchat/android/ui/MediaSendingManager.kt

@@ -739,7 +739,16 @@ class MediaSendingManager(
     fun handleTransferProgressEvent(evt: com.bitchat.android.mesh.TransferProgressEvent) {
         val msgId = synchronized(transferMessageMap) { transferMessageMap[evt.transferId] }
         if (msgId != null) {
-            if (evt.completed) {
+            if (evt.failed) {
+                messageManager.updateMessageDeliveryStatus(
+                    msgId,
+                    com.bitchat.android.model.DeliveryStatus.Failed("transfer could not be sent")
+                )
+                synchronized(transferMessageMap) {
+                    val msgIdRemoved = transferMessageMap.remove(evt.transferId)
+                    if (msgIdRemoved != null) messageTransferMap.remove(msgIdRemoved)
+                }
+            } else if (evt.completed) {
                 messageManager.updateMessageDeliveryStatus(
                     msgId,
                     com.bitchat.android.model.DeliveryStatus.Delivered(to = "mesh", at = java.util.Date())

app/src/test/java/com/bitchat/android/mesh/FragmentingPacketSenderTest.kt

@@ -0,0 +1,94 @@
+package com.bitchat.android.mesh
+
+import com.bitchat.android.model.RoutedPacket
+import com.bitchat.android.protocol.BitchatPacket
+import com.bitchat.android.protocol.MessageType
+import com.bitchat.android.util.AppConstants
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import org.junit.Assert.assertFalse
+import org.junit.Assert.assertTrue
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.robolectric.RobolectricTestRunner
+import java.util.Random
+
+@RunWith(RobolectricTestRunner::class)
+class FragmentingPacketSenderTest {
+
+    private val senderID = "1122334455667788"
+
+    private fun packetWithPayload(bytes: Int): BitchatPacket {
+        val payload = ByteArray(bytes)
+        Random(42).nextBytes(payload)
+        return BitchatPacket(
+            version = 2u,
+            type = MessageType.FILE_TRANSFER.value,
+            senderID = MeshPacketUtils.hexStringToByteArray(senderID),
+            recipientID = null,
+            timestamp = System.currentTimeMillis().toULong(),
+            payload = payload,
+            signature = null,
+            ttl = 7u
+        )
+    }
+
+    @Test
+    fun `oversized packet exceeding receiver fragment cap is rejected with fail event`() = runBlocking {
+        val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
+        val sender = FragmentingPacketSender(scope, FragmentManager(), "test")
+        // ~256 * 469 bytes fit; 1 MiB clearly exceeds MAX_FRAGMENTS_PER_ID
+        val packet = packetWithPayload(1024 * 1024)
+        var sent = false
+
+        val failed = java.util.concurrent.ConcurrentLinkedQueue<String>()
+        val collectJob = launch(Dispatchers.Default) {
+            TransferProgressManager.events.collect { event ->
+                if (event.failed) failed.add(event.transferId)
+            }
+        }
+        kotlinx.coroutines.delay(100) // activate subscription before emitting
+
+        val accepted = sender.send(RoutedPacket(packet, transferId = "oversize-test"), "test") { sent = true; true }
+        assertFalse(accepted)
+        assertFalse(sent)
+        withTimeout(5_000) {
+            while (!failed.contains("oversize-test")) {
+                kotlinx.coroutines.delay(10)
+            }
+        }
+        collectJob.cancel()
+        Unit
+    }
+
+    @Test
+    fun `packet within fragment cap is accepted`() = runBlocking {
+        val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
+        val sender = FragmentingPacketSender(scope, FragmentManager(), "test", interFragmentDelayMs = 0L)
+        val packet = packetWithPayload(10_000)
+        var writes = 0
+
+        val accepted = sender.send(RoutedPacket(packet, transferId = "fits-test"), "test") { writes += 1; true }
+        assertTrue(accepted)
+        withTimeout(5_000) {
+            while (writes == 0) {
+                kotlinx.coroutines.delay(10)
+            }
+        }
+        assertTrue(writes > 0)
+    }
+
+    @Test
+    fun `fragment count at cap boundary is not rejected`() {
+        val manager = FragmentManager()
+        val packet = packetWithPayload(AppConstants.Fragmentation.MAX_FRAGMENTS_PER_ID * 400)
+        val fragments = manager.createFragments(packet, AppConstants.Fragmentation.MAX_FRAGMENTS_PER_ID)
+        assertTrue(fragments.isNotEmpty())
+        assertTrue(fragments.size <= AppConstants.Fragmentation.MAX_FRAGMENTS_PER_ID)
+    }
+}

app/src/test/kotlin/com/bitchat/android/mesh/FragmentingPacketSenderConfirmedSendTest.kt

@@ -12,7 +12,7 @@ import org.junit.Assert.assertTrue
 import org.junit.Test
 
 @OptIn(ExperimentalCoroutinesApi::class)
-class FragmentingPacketSenderTest {
+class FragmentingPacketSenderConfirmedSendTest {
     @Test
     fun confirmedSendStaysPendingWhenExactRouteDisappearsBetweenFragments() = runTest {
         val fragments = listOf(packet(1), packet(2), packet(3))

docs/release-gate-runbook.md

@@ -242,3 +242,101 @@ command and stop the local relay/Tor fixture.
   not waive a mandatory scenario.
 - A flaky result is a failure until its cause is understood. Never average
   retries into a pass.
+
+## Appendix: mesh lab (ADB test hooks, debug builds)
+
+For day-to-day development there is a lighter-weight harness that drives a
+debug-only broadcast receiver (`app/src/debug/`, never shipped in release)
+exposing mesh operations over ADB: scan, connect, Noise handshake, DMs,
+public broadcast, announce, file send/receive, BLE toggle, state dumps, and
+raw packet injection. Results are JSON files in the app sandbox polled by the
+host (`cache/testhook/results/<id>.json`, also logged under tag `TestHook`).
+
+### Prerequisites
+
+- A JDK (e.g. the one bundled with Android Studio; set `JAVA_HOME`) and the
+  Android SDK platform-tools. `adb` must be on `PATH` or `ANDROID_HOME` set.
+- Python 3.10+ on the host. No third-party packages are required.
+- **Two physical Android devices** (API 26+, BLE) with USB debugging enabled,
+  both plugged into the host. Emulators are not supported (BLE mesh).
+- Verify both are visible: `adb devices` → note the serials.
+
+### Device preparation (important)
+
+Keep both phones **unlocked with the screen on** for the whole run. A locked
+or dozing device forces the app into POWER_SAVER (1 s BLE scan per 60 s),
+which makes discovery and handshakes take minutes and will flake every
+scenario. The harness runs `wake()` (dismiss keyguard, stretch screen
+timeout) during `setup`, but it cannot defeat a secure lock screen — unlock
+the devices manually first. Note that `svc power stayon` only helps while a
+device is actually charging.
+
+### Build and set up
+
+```sh
+./gradlew assembleDebug
+python3 tools/release_gate/mesh_lab.py setup \
+  --serial-a <serial-1> --serial-b <serial-2> \
+  --apk app/build/outputs/apk/debug/app-arm64-v8a-debug.apk
+```
+
+`setup` cycles Bluetooth, installs the APK, clears app data, grants all
+runtime permissions, wakes and launches the app, sets deterministic nicknames
+(`alice`/`bob`), and waits for mutual peer discovery. It is safe (and
+recommended) to rerun `setup` before each scenario batch; `--apk` may be
+omitted if the current build is already installed.
+
+### Run scenarios
+
+```sh
+python3 tools/release_gate/mesh_lab.py scenario all \
+  --serial-a <serial-1> --serial-b <serial-2> --out /tmp/meshlab-evidence
+```
+
+| Scenario | What it asserts |
+|---|---|
+| `dm` | Noise handshake both ways, encrypted DM round trips with content match |
+| `broadcast` | public mesh message A→B |
+| `file` | 1 KB broadcast file, receiver SHA-256 matches fixture |
+| `file_oversize` | >256-fragment broadcast file is rejected sender-side, receiver sees nothing |
+| `file_private` | Noise-encrypted private file, digest match |
+| `raw` | raw packet injection is accepted by the mesh |
+| `session_recovery` | force-stop B mid-session: identity persists, re-handshake, DMs flow again |
+| `identity_reset` | pm clear B mid-session: new identity, rediscovery, handshake, DMs |
+| `all` | every scenario above in sequence |
+
+Each run writes `<scenario>-evidence.json` to `--out` (digests, timings,
+session states, logcat excerpts on failure) and exits non-zero on failure.
+Evidence is a local diagnostic artifact; it may contain lab peer IDs and is
+not privacy-checked like release-gate bundles — do not publish it.
+
+### Ad-hoc commands
+
+Any hook command can be sent to one device directly:
+
+```sh
+python3 tools/release_gate/mesh_lab.py cmd --serial <serial> scan --extra timeout_ms=30000
+python3 tools/release_gate/mesh_lab.py cmd --serial <serial> handshake --extra peer=<peer-id>
+python3 tools/release_gate/mesh_lab.py cmd --serial <serial> state   # full mesh dump
+```
+
+See `TestHookDriver.kt` for the full command set (`ping`, `start`, `stop`,
+`whoami`, `set_nickname`, `scan`, `peers`, `connect`, `handshake`, `session`,
+`announce`, `broadcast_msg`, `dm_send`, `dm_recv`, `msg_recv`, `file_send`,
+`file_recv`, `file_cancel`, `raw_send`, `ble`, `state`, `clear_results`).
+
+### Troubleshooting
+
+- **Discovery/handshake timeouts**: almost always a locked or dozing phone —
+  unlock both devices and rerun `setup`. `cmd ... state` shows
+  `App In Background: true` and the BLE duty cycle when this is the cause.
+- **Stale app state after many churn runs**: `svc bluetooth disable/enable`
+  on both devices (done automatically by `setup`) clears zombie GATT links.
+- **Watch the wire**: `adb -s <serial> logcat -s TestHook MessageHandler
+  FragmentManager BitchatFilePacket` shows commands, results, decrypt
+  failures, fragment rejects, and saved incoming files in real time.
+- Results also persist on-device at
+  `run-as com.bitchat.droid cat cache/testhook/results/<id>.json`.
+
+Unlike the release gate, this harness is a development aid: it prints raw
+diagnostics and does not produce a privacy-checked approval bundle.

tools/release_gate/mesh_lab.py

@@ -0,0 +1,639 @@
+#!/usr/bin/env python3
+"""ADB-driven mesh test orchestrator for two (or more) live devices.
+
+Drives the debug-only TestHookReceiver in the app
+(intent action: com.bitchat.droid.TEST_HOOK) to perform mesh operations:
+peer scanning, connect, Noise handshake, DMs, file transfer, broadcast,
+announce, and raw packet injection.
+
+Each on-device command writes a JSON result to
+cache/testhook/results/<id>.json inside the app sandbox; this module polls
+for it via `run-as` and returns the parsed dict.
+
+Typical usage:
+    python3 tools/release_gate/mesh_lab.py setup --serial-a X --serial-b Y --apk app/build/outputs/apk/debug/app-debug.apk
+    python3 tools/release_gate/mesh_lab.py scenario dm --serial-a X --serial-b Y
+    python3 tools/release_gate/mesh_lab.py scenario all --serial-a X --serial-b Y
+    python3 tools/release_gate/mesh_lab.py cmd --serial X scan --extra timeout_ms=30000
+"""
+
+from __future__ import annotations
+
+import argparse
+import concurrent.futures
+import hashlib
+import json
+import random
+import shlex
+import subprocess
+import sys
+import tempfile
+import time
+import uuid
+from pathlib import Path
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[2]
+if str(REPOSITORY_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPOSITORY_ROOT))
+
+from tools.release_gate.android_lab import APPLICATION_ID, find_adb, run_adb
+
+TEST_HOOK_ACTION = "com.bitchat.droid.TEST_HOOK"
+TEST_HOOK_COMPONENT = f"{APPLICATION_ID}/com.bitchat.android.testhook.TestHookReceiver"
+RESULTS_DIR = "cache/testhook/results"
+DEVICE_TMP_DIR = "/data/local/tmp/meshlab"
+APP_FIXTURE_DIR = f"/data/data/{APPLICATION_ID}/cache/fixtures"
+
+PERMISSIONS = [
+    "android.permission.BLUETOOTH_SCAN",
+    "android.permission.BLUETOOTH_CONNECT",
+    "android.permission.BLUETOOTH_ADVERTISE",
+    "android.permission.ACCESS_FINE_LOCATION",
+    "android.permission.ACCESS_COARSE_LOCATION",
+    "android.permission.POST_NOTIFICATIONS",
+    "android.permission.NEARBY_WIFI_DEVICES",
+    "android.permission.RECORD_AUDIO",
+]
+
+
+class MeshLabError(Exception):
+    pass
+
+
+def _shell(serial: str, command: str) -> str:
+    return run_adb(serial, ["shell", command])
+
+
+class Device:
+    """One ADB-connected phone running a debug build with the test hook."""
+
+    def __init__(self, serial: str, alias: str):
+        self.serial = serial
+        self.alias = alias
+
+    # -- app lifecycle ------------------------------------------------------
+
+    def install(self, apk: Path) -> None:
+        result = subprocess.run(
+            [find_adb(), "-s", self.serial, "install", "-r", "-g", str(apk)],
+            check=False, capture_output=True, text=True, timeout=300,
+        )
+        if result.returncode != 0 or "Success" not in result.stdout:
+            raise MeshLabError(f"[{self.alias}] install failed: {result.stdout} {result.stderr}")
+
+    def grant_permissions(self) -> None:
+        for perm in PERMISSIONS:
+            subprocess.run(
+                [find_adb(), "-s", self.serial, "shell", "pm", "grant", APPLICATION_ID, perm],
+                check=False, capture_output=True, text=True, timeout=30,
+            )
+
+    def clear_app_data(self) -> None:
+        _shell(self.serial, f"am force-stop {APPLICATION_ID}")
+        output = _shell(self.serial, f"pm clear {APPLICATION_ID}")
+        if "Success" not in output:
+            raise MeshLabError(f"[{self.alias}] pm clear failed: {output}")
+
+    def force_stop(self) -> None:
+        _shell(self.serial, f"am force-stop {APPLICATION_ID}")
+
+    def launch(self) -> None:
+        _shell(self.serial, f"monkey -p {APPLICATION_ID} -c android.intent.category.LAUNCHER 1")
+        time.sleep(3)
+
+    def wake(self) -> None:
+        """Keep the screen on and the app foregrounded (full-power BLE duty cycle).
+
+        A backgrounded app drops to POWER_SAVER (1 s scan per 60 s), which makes
+        mesh reformation after restarts take minutes and scenarios flaky.
+        `svc power stayon` only applies while charging, so also stretch the
+        screen timeout as a fallback.
+        """
+        _shell(self.serial, "svc power stayon true")
+        _shell(self.serial, "settings put system screen_off_timeout 600000")
+        subprocess.run(
+            [find_adb(), "-s", self.serial, "shell", "locksettings", "set-disabled", "true"],
+            check=False, capture_output=True, text=True, timeout=30,
+        )
+        _shell(self.serial, "input keyevent KEYCODE_WAKEUP")
+        _shell(self.serial, "wm dismiss-keyguard")
+        _shell(self.serial, "input keyevent 82")  # dismiss non-secure keyguard
+        _shell(self.serial, "input swipe 500 1500 500 400")  # swipe-up dismiss
+
+    def reset_bluetooth(self) -> None:
+        """Cycle the BT adapter; clears zombie GATT connections from peer restarts."""
+        _shell(self.serial, "svc bluetooth disable")
+        time.sleep(2)
+        _shell(self.serial, "svc bluetooth enable")
+        time.sleep(3)
+
+    def enable_bluetooth(self) -> None:
+        subprocess.run(
+            [find_adb(), "-s", self.serial, "shell", "svc", "bluetooth", "enable"],
+            check=False, capture_output=True, text=True, timeout=30,
+        )
+
+    # -- fixtures -----------------------------------------------------------
+
+    def push_fixture(self, local: Path, name: str | None = None) -> str:
+        """Stage a fixture inside the app sandbox and return its app-readable path.
+
+        adb push lands files as shell:ext_data_rw, which the app cannot read
+        through the FUSE Android/data mount, so the bytes are piped through
+        the shell into the app's own cache directory via run-as.
+        """
+        fname = name or local.name
+        tmp = f"{DEVICE_TMP_DIR}/{fname}"
+        _shell(self.serial, f"mkdir -p {DEVICE_TMP_DIR}")
+        result = subprocess.run(
+            [find_adb(), "-s", self.serial, "push", str(local), tmp],
+            check=False, capture_output=True, text=True, timeout=120,
+        )
+        if result.returncode != 0:
+            raise MeshLabError(f"[{self.alias}] push failed: {result.stderr}")
+        target = f"{APP_FIXTURE_DIR}/{fname}"
+        _shell(
+            self.serial,
+            f"run-as {APPLICATION_ID} mkdir -p {APP_FIXTURE_DIR} && "
+            f"cat {tmp} | run-as {APPLICATION_ID} sh -c 'cat > {target}' && rm -f {tmp}",
+        )
+        return target
+
+    def clear_incoming(self) -> None:
+        _shell(
+            self.serial,
+            f"run-as {APPLICATION_ID} rm -rf cache/files/incoming cache/images/incoming",
+        )
+
+    # -- test hook commands -------------------------------------------------
+
+    def cmd(self, cmd: str, timeout_ms: int = 60_000, **extras: object) -> dict:
+        """Send a test-hook command and poll for its JSON result."""
+        cmd_id = uuid.uuid4().hex[:12]
+        _shell(self.serial, f"run-as {APPLICATION_ID} rm -f {RESULTS_DIR}/{cmd_id}.json")
+
+        args = [
+            "am", "broadcast", "-a", TEST_HOOK_ACTION,
+            "-n", TEST_HOOK_COMPONENT,
+            "--es", "cmd", cmd,
+            "--es", "id", cmd_id,
+            "--el", "timeout_ms", str(timeout_ms),
+            "--el", "overall_timeout_ms", str(timeout_ms + 30_000),
+        ]
+        for key, value in extras.items():
+            if value is None:
+                continue
+            if isinstance(value, bool):
+                args += ["--ez", key, "true" if value else "false"]
+            elif isinstance(value, int):
+                args += ["--el", key, str(value)]
+            else:
+                args += ["--es", key, str(value)]
+        try:
+            _shell(self.serial, " ".join(shlex.quote(a) for a in args))
+        except Exception as error:
+            # The shell occasionally hangs even though the broadcast was delivered;
+            # fall through to result polling, which is the authoritative channel.
+            print(f"[{self.alias}] warning: broadcast send for '{cmd}' raised: {error}", file=sys.stderr)
+
+        deadline = time.monotonic() + (timeout_ms + 60_000) / 1000
+        while time.monotonic() < deadline:
+            try:
+                raw = _shell(self.serial, f"run-as {APPLICATION_ID} cat {RESULTS_DIR}/{cmd_id}.json")
+                if raw.strip().startswith("{"):
+                    return json.loads(raw)
+            except Exception:
+                pass
+            time.sleep(1.0)
+        raise MeshLabError(f"[{self.alias}] timed out waiting for result of '{cmd}' ({cmd_id})")
+
+    def cmd_ok(self, cmd: str, timeout_ms: int = 60_000, **extras: object) -> dict:
+        result = self.cmd(cmd, timeout_ms=timeout_ms, **extras)
+        if result.get("status") != "ok":
+            raise MeshLabError(f"[{self.alias}] '{cmd}' failed: {result}")
+        return result
+
+    def logcat_dump(self, lines: int = 200) -> str:
+        return _shell(self.serial, f"logcat -d -t {lines}")
+
+
+# MARK: - fixtures
+
+FIXTURE_SIZES = {
+    "small_1k.bin": 1_024,
+    "medium_512k.bin": 512 * 1_024,
+    "large_2m.bin": 2 * 1_024 * 1_024,
+}
+
+
+def make_fixtures(directory: Path, seed: int = 1337, names: list[str] | None = None) -> dict[str, dict]:
+    directory.mkdir(parents=True, exist_ok=True)
+    fixtures = {}
+    rng = random.Random(seed)
+    for name, size in FIXTURE_SIZES.items():
+        if names is not None and name not in names:
+            rng.randbytes(size)  # keep the stream deterministic across subsets
+            continue
+        path = directory / name
+        data = rng.randbytes(size)
+        path.write_bytes(data)
+        fixtures[name] = {"path": path, "sha256": hashlib.sha256(data).hexdigest(), "bytes": size}
+    return fixtures
+
+
+# MARK: - setup
+
+def setup_pair(a: Device, b: Device, apk: Path | None, nickname_a: str, nickname_b: str) -> None:
+    for device, nickname in ((a, nickname_a), (b, nickname_b)):
+        device.reset_bluetooth()
+        device.enable_bluetooth()
+        if apk is not None:
+            device.install(apk)
+        device.clear_app_data()
+        device.grant_permissions()
+        device.wake()
+        device.launch()
+        device.cmd_ok("start")
+        device.cmd_ok("set_nickname", name=nickname)
+    wait_for_mutual_discovery(a, b)
+
+
+def whoami(device: Device) -> dict:
+    return device.cmd_ok("whoami")
+
+
+def wait_for_peer(device: Device, peer_id: str, timeout_s: int = 90) -> dict:
+    deadline = time.monotonic() + timeout_s
+    while time.monotonic() < deadline:
+        result = device.cmd_ok("peers")
+        for peer in result.get("peers", []):
+            if peer.get("id") == peer_id:
+                return peer
+        device.cmd_ok("announce")
+        time.sleep(3)
+    raise MeshLabError(f"[{device.alias}] peer {peer_id} not discovered within {timeout_s}s")
+
+
+def wait_for_mutual_discovery(a: Device, b: Device) -> None:
+    id_a = whoami(a)["peer_id"]
+    id_b = whoami(b)["peer_id"]
+    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
+        fa = pool.submit(wait_for_peer, a, id_b)
+        fb = pool.submit(wait_for_peer, b, id_a)
+        fa.result()
+        fb.result()
+
+
+# MARK: - scenarios
+
+def scenario_dm(a: Device, b: Device) -> dict:
+    """Handshake, then exchange DMs in both directions with content assertions."""
+    id_a = whoami(a)["peer_id"]
+    id_b = whoami(b)["peer_id"]
+
+    hs = a.cmd_ok("handshake", timeout_ms=60_000, peer=id_b)
+    hs_back = b.cmd_ok("handshake", timeout_ms=60_000, peer=id_a)
+
+    token_ab = f"dm-{uuid.uuid4().hex[:8]}"
+    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
+        recv = pool.submit(b.cmd_ok, "dm_recv", 60_000, peer=id_a, contains=token_ab)
+        time.sleep(2)
+        send = pool.submit(a.cmd_ok, "dm_send", 30_000, peer=id_b, content=f"hello b {token_ab}")
+        recv_result, send_result = recv.result(), send.result()
+    assert token_ab in recv_result["content"], recv_result
+
+    token_ba = f"dm-{uuid.uuid4().hex[:8]}"
+    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
+        recv = pool.submit(a.cmd_ok, "dm_recv", 60_000, peer=id_b, contains=token_ba)
+        time.sleep(2)
+        send = pool.submit(b.cmd_ok, "dm_send", 30_000, peer=id_a, content=f"hello a {token_ba}")
+        recv_result2, send_result2 = recv.result(), send.result()
+    assert token_ba in recv_result2["content"], recv_result2
+
+    return {
+        "handshake_a_to_b": hs, "handshake_b_to_a": hs_back,
+        "a_to_b": {"send": send_result, "recv": recv_result},
+        "b_to_a": {"send": send_result2, "recv": recv_result2},
+    }
+
+
+def scenario_broadcast(a: Device, b: Device) -> dict:
+    """Public broadcast from A received by B."""
+    id_a = whoami(a)["peer_id"]
+    token = f"bc-{uuid.uuid4().hex[:8]}"
+    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
+        recv = pool.submit(b.cmd_ok, "msg_recv", 60_000, contains=token)
+        time.sleep(2)
+        send = pool.submit(a.cmd_ok, "broadcast_msg", 30_000, content=f"broadcast {token}")
+        recv_result, send_result = recv.result(), send.result()
+    assert recv_result["from"] == id_a, recv_result
+    return {"send": send_result, "recv": recv_result}
+
+
+def scenario_file(a: Device, b: Device, fixtures: dict[str, dict], private: bool = False) -> dict:
+    """File transfer A -> B with sha256 integrity verification."""
+    id_b = whoami(b)["peer_id"]
+    b.clear_incoming()  # avoid name-uniquified collisions across runs
+    results = {}
+    for name, fixture in fixtures.items():
+        remote = a.push_fixture(fixture["path"])
+        send_kwargs: dict[str, object] = {"path": remote}
+        if private:
+            send_kwargs["peer"] = id_b
+        with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
+            recv = pool.submit(b.cmd_ok, "file_recv", 240_000, name_contains=name)
+            time.sleep(2)
+            send = pool.submit(a.cmd_ok, "file_send", 240_000, **send_kwargs)
+            recv_result, send_result = recv.result(), send.result()
+        digest_ok = recv_result["sha256"] == fixture["sha256"]
+        results[name] = {
+            "send": send_result, "recv": recv_result,
+            "expected_sha256": fixture["sha256"], "digest_match": digest_ok,
+        }
+        if not digest_ok:
+            raise MeshLabError(
+                f"file '{name}' digest mismatch: {recv_result['sha256']} != {fixture['sha256']}"
+            )
+    return results
+
+
+def scenario_raw(a: Device, b: Device) -> dict:
+    """Raw packet injection (unsigned announce-type packet) reaches the mesh."""
+    payload = b"meshlab-raw-" + uuid.uuid4().hex[:8].encode()
+    result = a.cmd_ok("raw_send", 30_000, type="05", payload_hex=payload.hex())
+    return {"send": result}
+
+
+# MARK: - session / identity churn scenarios
+
+def _dm_roundtrip(a: Device, b: Device, id_a: str, id_b: str) -> dict:
+    """Exchange DMs in both directions with content assertions."""
+    token_ab = f"dm-{uuid.uuid4().hex[:8]}"
+    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
+        recv = pool.submit(b.cmd_ok, "dm_recv", 60_000, peer=id_a, contains=token_ab)
+        time.sleep(2)
+        send = pool.submit(a.cmd_ok, "dm_send", 30_000, peer=id_b, content=f"hello b {token_ab}")
+        recv_ab, send_ab = recv.result(), send.result()
+    assert token_ab in recv_ab["content"], recv_ab
+
+    token_ba = f"dm-{uuid.uuid4().hex[:8]}"
+    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
+        recv = pool.submit(a.cmd_ok, "dm_recv", 60_000, peer=id_b, contains=token_ba)
+        time.sleep(2)
+        send = pool.submit(b.cmd_ok, "dm_send", 30_000, peer=id_a, content=f"hello a {token_ba}")
+        recv_ba, send_ba = recv.result(), send.result()
+    assert token_ba in recv_ba["content"], recv_ba
+    return {"a_to_b": recv_ab, "b_to_a": recv_ba}
+
+
+def wait_session_established(device: Device, peer_id: str, timeout_s: int = 90) -> dict:
+    deadline = time.monotonic() + timeout_s
+    last: dict = {}
+    while time.monotonic() < deadline:
+        last = device.cmd_ok("session", peer=peer_id)
+        if last.get("established"):
+            return last
+        time.sleep(2)
+    raise MeshLabError(
+        f"[{device.alias}] session with {peer_id} not established within {timeout_s}s (last: {last})"
+    )
+
+
+def ensure_direct_link(a: Device, b: Device, id_a: str, id_b: str) -> None:
+    """Wait for rediscovery, then force a direct GATT connection both ways.
+
+    Backgrounded devices drop to POWER_SAVER duty cycles (1 s scan per 60 s), so
+    passively waiting for the mesh to reform takes minutes. The explicit connect
+    makes restart scenarios deterministic.
+    """
+    wait_for_peer(a, id_b, timeout_s=120)
+    wait_for_peer(b, id_a, timeout_s=120)
+    for device, peer in ((a, id_b), (b, id_a)):
+        result = device.cmd("connect", timeout_ms=45_000, peer=peer)
+        if result.get("status") == "ok" and result.get("direct"):
+            continue
+        # Already acceptable if the mesh formed a direct link on its own.
+        peers = device.cmd_ok("peers").get("peers", [])
+        match = next((p for p in peers if p.get("id") == peer), None)
+        if not match or not match.get("direct"):
+            raise MeshLabError(f"[{device.alias}] no direct link to {peer}: connect={result}")
+
+
+def force_handshake(device: Device, peer_id: str, attempts: int = 5, per_attempt_s: int = 20) -> dict:
+    """Retry explicit handshakes; inits can be lost while links settle."""
+    last: dict = {}
+    for _ in range(attempts):
+        last = device.cmd("handshake", timeout_ms=per_attempt_s * 1000, peer=peer_id)
+        if last.get("status") == "ok":
+            return last
+        time.sleep(2)
+    raise MeshLabError(f"[{device.alias}] handshake with {peer_id} failed after {attempts} attempts (last: {last})")
+
+
+def scenario_session_recovery(a: Device, b: Device) -> dict:
+    """Process death on B: identity must persist, in-memory Noise sessions are lost.
+
+    Expected recovery flow: A's DM sent with its stale session is dropped by B
+    (B has no session and no kick path on pure decrypt failure); B's outgoing DM
+    auto-triggers a fresh handshake; subsequent DMs must flow both ways.
+    """
+    id_a = whoami(a)["peer_id"]
+    id_b = whoami(b)["peer_id"]
+    a.cmd_ok("handshake", 60_000, peer=id_b)
+    b.cmd_ok("handshake", 60_000, peer=id_a)
+    baseline = _dm_roundtrip(a, b, id_a, id_b)
+
+    b.force_stop()
+    b.wake()
+    b.launch()
+    b.cmd_ok("start")
+    b.cmd_ok("set_nickname", name="bob")
+    id_b_after = whoami(b)["peer_id"]
+    if id_b_after != id_b:
+        raise MeshLabError(f"identity changed across process death: {id_b} -> {id_b_after}")
+
+    wait_for_peer(a, id_b, timeout_s=120)
+    wait_for_peer(b, id_a, timeout_s=120)
+    ensure_direct_link(a, b, id_a, id_b)
+
+    # A -> B with A's stale session: B lost its in-memory session; drop expected.
+    a.cmd_ok("dm_send", 30_000, peer=id_b, content=f"stale-{uuid.uuid4().hex[:8]}")
+    # B -> A: no session on B, sendPrivateMessage auto-fires the re-handshake.
+    # The fire-and-forget handshake has no retry, so repeat the trigger, then
+    # fall back to explicit handshake commands if the auto-path stays stuck.
+    session_b: dict = {}
+    for _attempt in range(3):
+        b.cmd_ok("dm_send", 30_000, peer=id_a, content=f"trigger-{uuid.uuid4().hex[:8]}")
+        try:
+            session_b = wait_session_established(b, id_a, timeout_s=20)
+            break
+        except MeshLabError:
+            continue
+    if not session_b:
+        force_handshake(b, id_a)
+        session_b = wait_session_established(b, id_a, timeout_s=30)
+
+    session_a = wait_session_established(a, id_b)
+    recovered = _dm_roundtrip(a, b, id_a, id_b)
+    return {
+        "identity_preserved": True,
+        "baseline": baseline,
+        "session_a": session_a,
+        "session_b": session_b,
+        "recovered": recovered,
+    }
+
+
+def scenario_identity_reset(a: Device, b: Device) -> dict:
+    """pm clear on B mid-session: new identity, rediscovery, fresh handshake and DMs."""
+    id_a = whoami(a)["peer_id"]
+    id_b_old = whoami(b)["peer_id"]
+    a.cmd_ok("handshake", 60_000, peer=id_b_old)
+    b.cmd_ok("handshake", 60_000, peer=id_a)
+    _dm_roundtrip(a, b, id_a, id_b_old)
+
+    b.clear_app_data()
+    b.grant_permissions()
+    b.wake()
+    b.launch()
+    b.cmd_ok("start")
+    b.cmd_ok("set_nickname", name="bob")
+    id_b_new = whoami(b)["peer_id"]
+    if id_b_new == id_b_old:
+        raise MeshLabError("identity survived pm clear")
+
+    wait_for_peer(a, id_b_new, timeout_s=180)
+    ensure_direct_link(a, b, id_a, id_b_new)
+    force_handshake(a, id_b_new)
+    force_handshake(b, id_a)
+    recovered = _dm_roundtrip(a, b, id_a, id_b_new)
+
+    # Inspect how A treats the dead peer's stale session (evidence, not an assertion).
+    stale = a.cmd("session", peer=id_b_old)
+    return {
+        "old_peer_id": id_b_old,
+        "new_peer_id": id_b_new,
+        "identity_changed": True,
+        "recovered": recovered,
+        "stale_session_on_a": stale,
+    }
+
+
+def scenario_file_oversize(a: Device, b: Device, fixtures: dict[str, dict]) -> dict:
+    """Oversized broadcast file must be rejected sender-side (>256 fragments)."""
+    fixture = fixtures["medium_512k.bin"]
+    remote = a.push_fixture(fixture["path"])
+    send = a.cmd("file_send", timeout_ms=60_000, path=remote)
+    rejected = send.get("status") == "error" and "rejected" in send.get("error", "")
+    if not rejected:
+        raise MeshLabError(f"expected sender-side rejection, got: {send}")
+    # Receiver must not see any file appear.
+    recv = b.cmd("file_recv", timeout_ms=15_000, name_contains="medium_512k")
+    if recv.get("status") == "ok":
+        raise MeshLabError(f"receiver unexpectedly saved an oversized file: {recv}")
+    return {"send": send, "receiver_saw_file": False}
+
+
+SCENARIOS = {
+    "dm": scenario_dm,
+    "broadcast": scenario_broadcast,
+    # Broadcast transfers are receiver-capped at 256 fragments (~120 KB); only
+    # the small fixture is end-to-end receivable.
+    "file": lambda a, b: scenario_file(
+        a, b,
+        make_fixtures(Path(tempfile.mkdtemp(prefix="meshlab-fixtures-")), names=["small_1k.bin"]),
+    ),
+    "file_oversize": lambda a, b: scenario_file_oversize(
+        a, b, make_fixtures(Path(tempfile.mkdtemp(prefix="meshlab-fixtures-")))
+    ),
+    # Private media is hard-capped at 256 fragments (PrivateMediaTransfer), so only
+    # the small fixture fits; larger sizes are expected to be rejected by the sender.
+    "file_private": lambda a, b: scenario_file(
+        a, b,
+        make_fixtures(Path(tempfile.mkdtemp(prefix="meshlab-fixtures-")), names=["small_1k.bin"]),
+        private=True,
+    ),
+    "raw": scenario_raw,
+    "session_recovery": scenario_session_recovery,
+    "identity_reset": scenario_identity_reset,
+}
+
+
+def run_scenario(name: str, a: Device, b: Device, out: Path | None) -> dict:
+    started = time.time()
+    evidence: dict[str, object] = {"scenario": name, "devices": [a.alias, b.alias]}
+    try:
+        if name == "all":
+            evidence["results"] = {n: run_scenario(n, a, b, None)["results"] for n in SCENARIOS}
+        else:
+            evidence["results"] = SCENARIOS[name](a, b)
+        evidence["status"] = "pass"
+    except (MeshLabError, AssertionError) as error:
+        evidence["status"] = "fail"
+        evidence["error"] = str(error)
+        evidence["logcat"] = {d.alias: d.logcat_dump() for d in (a, b)}
+    evidence["duration_s"] = round(time.time() - started, 1)
+    if out is not None:
+        out.mkdir(parents=True, exist_ok=True)
+        (out / f"{name}-evidence.json").write_text(json.dumps(evidence, indent=2, default=str))
+    return evidence
+
+
+# MARK: - CLI
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    commands = parser.add_subparsers(dest="command", required=True)
+
+    setup = commands.add_parser("setup", help="install, grant, launch, nickname, discover")
+    setup.add_argument("--serial-a", required=True)
+    setup.add_argument("--serial-b", required=True)
+    setup.add_argument("--apk", type=Path, default=None)
+    setup.add_argument("--nickname-a", default="alice")
+    setup.add_argument("--nickname-b", default="bob")
+
+    scenario = commands.add_parser("scenario", help="run a test scenario on two devices")
+    scenario.add_argument("name", choices=[*SCENARIOS.keys(), "all"])
+    scenario.add_argument("--serial-a", required=True)
+    scenario.add_argument("--serial-b", required=True)
+    scenario.add_argument("--out", type=Path, default=None, help="evidence output directory")
+
+    raw = commands.add_parser("cmd", help="send a raw test-hook command to one device")
+    raw.add_argument("--serial", required=True)
+    raw.add_argument("cmd")
+    raw.add_argument("--extra", action="append", default=[], help="key=value extra (repeatable)")
+    raw.add_argument("--timeout-ms", type=int, default=60_000)
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = build_parser().parse_args(argv)
+    try:
+        if args.command == "setup":
+            setup_pair(
+                Device(args.serial_a, "alpha"), Device(args.serial_b, "beta"),
+                args.apk, args.nickname_a, args.nickname_b,
+            )
+            print(json.dumps({"status": "ok", "step": "setup"}))
+        elif args.command == "scenario":
+            evidence = run_scenario(
+                args.name, Device(args.serial_a, "alpha"), Device(args.serial_b, "beta"), args.out
+            )
+            print(json.dumps(evidence, indent=2, default=str))
+            return 0 if evidence["status"] == "pass" else 1
+        elif args.command == "cmd":
+            extras: dict[str, object] = {}
+            for item in args.extra:
+                key, _, value = item.partition("=")
+                extras[key] = int(value) if value.isdigit() else value
+            result = Device(args.serial, "device").cmd(args.cmd, timeout_ms=args.timeout_ms, **extras)
+            print(json.dumps(result, indent=2, default=str))
+            return 0 if result.get("status") == "ok" else 1
+        return 0
+    except MeshLabError as error:
+        print(f"mesh lab error: {error}", file=sys.stderr)
+        return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())