Make Android microphone capture stoppable and cleanup resilient

Use nonblocking AudioRecord reads with data and timestamp deadlines. Serialize idempotent close, finish interrupted joins before release, isolate callback failures, and abort stereo fallback when partial-record cleanup fails. Close failed shared readers to new subscriptions, roll back worker startup, isolate observers, preserve interrupt state, and release the last consumer exactly once. Add an independent capture-tests JVM module to the standard Android unit-test task. Validation: 38 capture scenarios (25 AudioRecord plus 13 PCM/PTT), 455 JVM tests, assembleDebug, core build, and full check.sh (225 tests plus proxy/burst/load) passed.
2026-10-28 00:53:01 +01:00
1 parent 74035fc043
commit 005d136dc7
@@ -79,7 +79,7 @@ android {

 // Зависимость только между задачами: Android fixtures из JVM-модуля не входят в classpath приложения/APK.
 tasks.withType<org.gradle.api.tasks.testing.Test>().configureEach {
-    dependsOn(":router-tests:test")
+    dependsOn(":router-tests:test", ":capture-tests:test")
    val messageLinkCases = rootProject.file("../chatgui/tests/message_links_cases.tsv")
    inputs.file(messageLinkCases)
    systemProperty("utun.messageLinkCases", messageLinkCases.absolutePath)
@@ -6,21 +6,36 @@ import android.media.AudioRecord
 import android.media.AudioTimestamp
 import android.media.MediaRecorder
 import android.os.Build
+import java.util.concurrent.locks.LockSupport

-/** Один аппаратный reader. Контроллер закрывает его перед сменой маршрута;
- * только подтверждённый свежий PCM попадает в общую временную шкалу. */
+/**
+ * Один AudioRecord, 48 kHz, кадры 20 мс stereo (mono дублируется в два канала).
+ * Владелец закрывает reader перед сменой маршрута, вне accept/failed/confirmed callbacks.
+ * NON_BLOCKING read позволяет остановиться и при отказе stop; отсутствие полного кадра ограничено deadline.
+ * Публикуется только PCM выбранного, не silenced входа со свежим monotonic timestamp.
+ * close сериализован и идемпотентен; прерванный join завершается до release, затем флаг interrupt восстанавливается.
+ */
 internal class AndroidMicrophoneCapture private constructor(
    private val record: AudioRecord, private val input: AudioDeviceInfo, private val generation: Long,
    private val accept: (MicrophoneCapture.Frame) -> Boolean, private val failed: (Exception) -> Unit,
-    private val confirmed: () -> Unit
+    private val confirmed: () -> Unit, private val nowNs: () -> Long
 ) : AutoCloseable {
    @Volatile private var running = true
    private val worker = Thread(::read, "media-microphone")
+    private val closeLock = Any()
+    private var stopRequested = false
+    private var released = false

    companion object {
+        /**
+         * Открыть выбранный input: voice принудительно mono/VOICE_COMMUNICATION, stereo разрешает fallback в mono.
+         * micSource: 0 auto, 1 MIC, 2 UNPROCESSED, 3 CAMCORDER. generation передаётся в каждый Frame без изменения.
+         * accept возвращает false при переполнении/старом PCM; confirmed вызывается один раз после первого принятого кадра.
+         * failed сообщает ошибку worker. nowNs использует ту же monotonic шкалу, что AudioTimestamp.
+         */
        fun open(input: AudioDeviceInfo, voice: Boolean, stereo: Boolean, micSource: Int, generation: Long,
                 accept: (MicrophoneCapture.Frame) -> Boolean, failed: (Exception) -> Unit,
-                 confirmed: () -> Unit): AndroidMicrophoneCapture {
+                 confirmed: () -> Unit, nowNs: () -> Long = System::nanoTime): AndroidMicrophoneCapture {
            val source = if (voice) MediaRecorder.AudioSource.VOICE_COMMUNICATION else when (micSource) {
                1 -> MediaRecorder.AudioSource.MIC
                2 -> MediaRecorder.AudioSource.UNPROCESSED
@@ -37,16 +52,21 @@ internal class AndroidMicrophoneCapture private constructor(
                    check(min > 0) { "Unsupported capture mask=$mask min=$min" }
                    record = AudioRecord(source, 48000, mask, AudioFormat.ENCODING_PCM_16BIT, maxOf(min * 2, 15360))
                    check(record.state == AudioRecord.STATE_INITIALIZED) { "AudioRecord not initialized mask=$mask" }
+                    check(record.channelCount in 1..2) { "Unsupported capture channels=${record.channelCount}" }
                    check(record.setPreferredDevice(input)) { "Input selection rejected id=${input.id}" }
                    record.startRecording()
                    check(record.recordingState == AudioRecord.RECORDSTATE_RECORDING) { "AudioRecord did not start" }
                    LogManager.addLog("INFO", "media_audio", "capture opened generation=$generation input=${input.id}/${input.type} " +
                        "name=${input.productName} source=$source channels=${record.channelCount} rate=48000")
-                    return AndroidMicrophoneCapture(record, input, generation, accept, failed, confirmed).also { it.worker.start() }
+                    return AndroidMicrophoneCapture(record, input, generation, accept, failed, confirmed, nowNs).also { it.worker.start() }
                } catch (e: Exception) {
                    failure = e
                    LogManager.addLog("WARN", "media_audio", "capture open failed input=${input.id} mask=$mask: $e")
-                    record?.release()
+                    try { record?.release() } catch (cleanup: Exception) {
+                        e.addSuppressed(cleanup)
+                        LogManager.addLog("ERROR", "media_audio", "capture open cleanup failed input=${input.id} mask=$mask: $cleanup")
+                        throw IllegalStateException("Cannot release failed capture input=${input.id}; fallback aborted", e)
+                    }
                }
            }
            throw IllegalStateException("Cannot open microphone ${input.productName}", failure)
@@ -54,25 +74,38 @@ internal class AndroidMicrophoneCapture private constructor(
    }

    private fun read() {
-        val stamp = AudioTimestamp()
-        val channels = record.channelCount
-        var total = 0L
-        var validSince = 0L
-        var ready = false
-        var lateSince = 0L
-        val deadline = System.nanoTime() + 5_000_000_000L
        try {
-            android.os.Process.setThreadPriority(android.os.Process.THREAD_PRIORITY_AUDIO)
+            val stamp = AudioTimestamp()
+            val channels = record.channelCount
+            var total = 0L
+            var validSince = 0L
+            var ready = false
+            var lateSince = 0L
+            var timestampMissingSince = 0L
+            val deadline = nowNs() + 5_000_000_000L
+            var lastFrameNs = nowNs()
+            try { android.os.Process.setThreadPriority(android.os.Process.THREAD_PRIORITY_AUDIO) } catch (e: Exception) {
+                LogManager.addLog("WARN", "media_audio", "capture audio priority unavailable input=${input.id}: $e")
+            }
            while (running) {
                val pcm = ShortArray(960 * channels)
                var used = 0
                val before = record.routedDevice?.id
                while (running && used < pcm.size) {
-                    val count = record.read(pcm, used, pcm.size - used, AudioRecord.READ_BLOCKING)
-                    check(count > 0 || !running) { "AudioRecord read=$count input=${input.id}" }
-                    if (count > 0) used += count
+                    check(!Thread.currentThread().isInterrupted) { "Capture reader interrupted input=${input.id}" }
+                    val now = nowNs()
+                    check(now < if (ready) lastFrameNs + 1_000_000_000L else deadline) {
+                        "Microphone data timeout input=${input.id} ready=$ready samples=$used/${pcm.size}"
+                    }
+                    val count = record.read(pcm, used, pcm.size - used, AudioRecord.READ_NON_BLOCKING)
+                    if (!running) break
+                    check(count >= 0 && count <= pcm.size - used && count % channels == 0) {
+                        "AudioRecord read=$count input=${input.id} remaining=${pcm.size - used} channels=$channels"
+                    }
+                    if (count == 0) LockSupport.parkNanos(this, 5_000_000L) else used += count
                }
                if (!running) break
+                lastFrameNs = nowNs()
                val first = total; total += 960
                val actual = record.routedDevice?.id
                val silenced = Build.VERSION.SDK_INT >= 29 && record.activeRecordingConfiguration?.isClientSilenced == true
@@ -80,20 +113,30 @@ internal class AndroidMicrophoneCapture private constructor(
                if (!valid) {
                    if (ready) error("Microphone route lost selected=${input.id} actual=$actual silenced=$silenced")
                    validSince = 0
-                } else if (validSince == 0L) validSince = System.nanoTime()
+                } else if (validSince == 0L) validSince = nowNs()
                val measured = record.getTimestamp(stamp, AudioTimestamp.TIMEBASE_MONOTONIC) == AudioRecord.SUCCESS
                val time = if (measured) stamp.nanoTime - (stamp.framePosition - first) * 1_000_000_000 / 48000 else 0L
-                if (!valid || !measured || time < validSince) {
-                    check(System.nanoTime() < deadline) {
-                        "Microphone confirmation timeout selected=${input.id} actual=$actual silenced=$silenced timestamp=$time"
+                val now = nowNs()
+                check(ready || now < deadline) { "Microphone confirmation timeout input=${input.id} now=$now deadline=$deadline" }
+                if (!valid || !measured || stamp.nanoTime <= 0 || stamp.framePosition < 0 || time < validSince || time > now) {
+                    if (timestampMissingSince == 0L) {
+                        timestampMissingSince = now
+                        LogManager.addLog(if (ready) "WARN" else "DEBUG", "media_audio",
+                            "capture waiting metadata input=${input.id} actual=$actual silenced=$silenced ready=$ready " +
+                                "measured=$measured position=${stamp.framePosition} timestamp=$time validSince=$validSince now=$now")
+                    }
+                    check(now < if (ready) timestampMissingSince + 1_000_000_000L else deadline) {
+                        "Microphone confirmation timeout selected=${input.id} actual=$actual silenced=$silenced " +
+                            "ready=$ready measured=$measured position=${stamp.framePosition} timestamp=$time now=$now"
                    }
                    continue
                }
+                timestampMissingSince = 0
                val stereo = if (channels == 2) pcm else ShortArray(1920) { pcm[it / 2] }
                val accepted = accept(MicrophoneCapture.Frame(stereo, time, first, generation = generation))
                if (!accepted) {
-                    if (lateSince == 0L) lateSince = System.nanoTime()
-                    check(System.nanoTime() - lateSince < 1_000_000_000L) { "Microphone PCM exceeds capture reserve input=${input.id}" }
+                    if (lateSince == 0L) lateSince = nowNs()
+                    check(nowNs() - lateSince < 1_000_000_000L) { "Microphone PCM exceeds capture reserve input=${input.id}" }
                    continue
                }
                lateSince = 0
@@ -105,18 +148,43 @@ internal class AndroidMicrophoneCapture private constructor(
            }
        } catch (e: Exception) {
            if (running) {
+                running = false
                LogManager.addLog("ERROR", "media_audio", "capture failed generation=$generation input=${input.id}: $e")
-                failed(e)
+                try { failed(e) } catch (callback: Exception) {
+                    LogManager.addLog("ERROR", "media_audio", "capture failure callback threw input=${input.id}: $callback")
+                }
            }
        }
    }

    override fun close() {
-        running = false
-        try { record.stop() }
-        catch (e: Exception) { LogManager.addLog("WARN", "media_audio", "capture stop input=${input.id}: $e") }
-        worker.join()
-        record.release()
-        LogManager.addLog("INFO", "media_audio", "capture released generation=$generation input=${input.id}")
+        check(Thread.currentThread() !== worker) {
+            LogManager.addLog("ERROR", "media_audio", "capture close from reader rejected input=${input.id}")
+            "Capture cannot join its own reader"
+        }
+        synchronized(closeLock) {
+            if (released) return
+            running = false
+            LockSupport.unpark(worker)
+            if (!stopRequested) {
+                stopRequested = true
+                try { record.stop() } catch (e: Exception) {
+                    LogManager.addLog("WARN", "media_audio", "capture stop input=${input.id}: $e")
+                }
+            }
+            var interrupted = false
+            while (worker.isAlive) try { worker.join() } catch (e: InterruptedException) {
+                interrupted = true
+                LogManager.addLog("WARN", "media_audio", "capture join interrupted input=${input.id}; finishing cleanup")
+            }
+            try {
+                record.release()
+                released = true
+                LogManager.addLog("INFO", "media_audio", "capture released generation=$generation input=${input.id}")
+            } catch (e: Exception) {
+                LogManager.addLog("ERROR", "media_audio", "capture release failed input=${input.id}; close can retry: $e")
+                throw e
+            } finally { if (interrupted) Thread.currentThread().interrupt() }
+        }
    }
 }
@@ -3,11 +3,18 @@ package com.utun.chat.data
 import java.util.concurrent.ArrayBlockingQueue
 import java.util.concurrent.TimeUnit

-/** Один аппаратный reader, независимые ограниченные очереди потребителей. PCM каждого потребителя — своя копия.
- * Закрытие подписки не останавливает соседей; последний владелец останавливает read, делает join и release.
- * PTT попадает в ту же очередь, что PCM, поэтому native TX всегда имеет одного последовательного владельца. */
+/**
+ * Один Device reader, независимые очереди потребителей; PCM каждого — своя копия.
+ * Последний close останавливает/прерывает reader, дожидается join, закрывает Device и уведомляет released.
+ * Device.stop или interrupt обязан разблокировать read. Подписка закрывается вне reader/callback read.
+ * closed запрещает новые подписки и после read failure; ресурсы освобождает последний владелец.
+ * PTT и PCM упорядочены одной очередью (32 места, минимум 8 зарезервированы для управления).
+ * changed/released — быстрые observers вне монитора; их исключения логируются и не прерывают cleanup.
+ * threadFactory возвращает новый незапущенный Thread. log должен завершаться быстро и не бросать исключений.
+ */
 class MicrophoneCapture(private val device: Device, private val log: (String, String) -> Unit, private val released: () -> Unit = {},
-                        private val changed: () -> Unit = {}) {
+                        private val changed: () -> Unit = {},
+                        private val threadFactory: (Runnable) -> Thread = { Thread(it, "microphone-capture") }) {
    interface Device {
        val channels: Int
        val inputId: Int?
@@ -49,46 +56,97 @@ class MicrophoneCapture(private val device: Device, private val log: (String, St
            return false
        }
        fun poll(): Event? {
+            if (!attached) return null
            failure?.let { throw IllegalStateException("Microphone read failed", it) }
-            return if (attached) events.poll(20, TimeUnit.MILLISECONDS) else null
+            val event = events.poll(20, TimeUnit.MILLISECONDS)
+            return if (attached) event else null
        }
        override fun close() {
+            check(Thread.currentThread() !== worker) {
+                log("ERROR", "capture subscription close from reader rejected consumer=$name")
+                "Capture cannot join its own reader"
+            }
            synchronized(this) { if (!attached) return; attached = false; events.clear() }
            detach(this)
        }
    }

-    @Synchronized fun subscribe(name: String): Subscription {
-        check(reading && failure == null) { "Microphone is closed or failed" }
+    fun subscribe(name: String): Subscription {
        val subscription = Subscription(name)
-        consumers.add(subscription)
-        if (worker == null) worker = Thread({
-            try {
-                while (reading) {
-                    val frame = device.read()
-                    val targets = synchronized(this) { consumers.toList() }
-                    for (target in targets) target.deliver(frame)
-                }
-            } catch (e: Exception) {
-                if (reading) { failure = e; log("ERROR", "capture reader failed: ${e.message}") }
+        var startupFailure: Throwable? = null
+        val count = synchronized(this) {
+            check(reading && failure == null) {
+                log("ERROR", "capture subscription rejected consumer=$name reading=$reading failure=$failure")
+                "Microphone is closed or failed"
            }
-        }, "microphone-capture").apply { start() }
-        log("INFO", "capture attached consumer=$name channels=$channels consumers=${consumers.size}")
-        changed()
+            consumers.add(subscription)
+            if (worker == null) try {
+                val next = threadFactory(Runnable(::read))
+                check(next.state == Thread.State.NEW) { "Capture requires a fresh unstarted worker" }
+                worker = next
+                next.start()
+            } catch (e: Throwable) {
+                failure = if (e is Exception) e else IllegalStateException("Capture worker startup failed", e)
+                reading = false; consumers.remove(subscription); startupFailure = e
+            }
+            consumers.size
+        }
+        startupFailure?.let {
+            log("ERROR", "capture worker startup failed consumer=$name: $it")
+            releaseDevice()
+            throw it
+        }
+        log("INFO", "capture attached consumer=$name channels=$channels consumers=$count")
+        notifyChanged()
        return subscription
    }

+    private fun read() {
+        try {
+            while (reading) {
+                val frame = device.read()
+                if (!reading) break
+                val targets = synchronized(this) { consumers.toList() }
+                for (target in targets) target.deliver(frame)
+            }
+        } catch (e: Exception) {
+            val unexpected = synchronized(this) {
+                reading.also { if (it) { failure = e; reading = false } }
+            }
+            if (unexpected) {
+                log("ERROR", "capture reader failed: $e")
+                notifyChanged()
+            }
+        }
+    }
+
+    private fun notifyChanged() {
+        try { changed() } catch (e: Exception) { log("ERROR", "capture changed observer failed: $e") }
+    }
+
    private fun detach(subscription: Subscription) {
        val last = synchronized(this) {
            consumers.remove(subscription)
            log("INFO", "capture detached consumer=${subscription.name} remaining=${consumers.size}")
            consumers.isEmpty().also { if (it) reading = false }
        }
-        changed()
+        notifyChanged()
        if (!last) return
-        try { device.stop() } catch (e: Exception) { log("WARN", "capture stop failed: ${e.message}") }
-        worker?.join()
-        try { device.close() } catch (e: Exception) { log("ERROR", "capture release failed: ${e.message}") }
-        finally { released() }
+        releaseDevice()
+    }
+
+    private fun releaseDevice() {
+        var interrupted = Thread.interrupted()
+        try {
+            try { device.stop() } catch (e: Exception) { log("WARN", "capture stop failed: $e") }
+            val thread = worker
+            thread?.interrupt()
+            while (thread?.isAlive == true) try { thread.join() } catch (e: InterruptedException) {
+                interrupted = true
+                log("WARN", "capture join interrupted; finishing cleanup")
+            }
+            try { device.close() } catch (e: Exception) { log("ERROR", "capture release failed: $e") }
+            try { released() } catch (e: Exception) { log("ERROR", "capture released observer failed: $e") }
+        } finally { if (interrupted) Thread.currentThread().interrupt() }
    }
 }
@@ -4,6 +4,9 @@ import org.junit.Assert.*
 import org.junit.Test
 import java.util.concurrent.LinkedBlockingQueue
 import java.util.concurrent.atomic.AtomicInteger
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.FutureTask
+import java.util.concurrent.TimeUnit

 class MicrophoneCaptureTest {
    private class Device : MicrophoneCapture.Device {
@@ -67,4 +70,150 @@ class MicrophoneCaptureTest {
            assertEquals(MicrophoneCapture.Ptt(true), radio.poll()); assertEquals(MicrophoneCapture.Ptt(false), radio.poll())
        } finally { radio.close(); stream.close() }
    }
+
+    private class FaultyDevice : MicrophoneCapture.Device {
+        override val channels = 2
+        override val inputId = 7
+        val input = LinkedBlockingQueue<Any>()
+        val entered = CountDownLatch(1)
+        val stopped = CountDownLatch(1)
+        val stops = AtomicInteger()
+        val closes = AtomicInteger()
+        var failStop = false
+        var failClose = false
+        var onRead: () -> Unit = {}
+        override fun read(): MicrophoneCapture.Frame {
+            entered.countDown()
+            val event = input.take()
+            onRead()
+            if (event is Exception) throw event
+            return event as MicrophoneCapture.Frame
+        }
+        override fun stop() {
+            stops.incrementAndGet(); stopped.countDown()
+            check(!failStop) { "device stop denied" }
+            input.offer(MicrophoneCapture.Frame(shortArrayOf(), 0, 0))
+        }
+        override fun close() { closes.incrementAndGet(); check(!failClose) { "device close denied" } }
+    }
+
+    private fun diagnostic(logs: LinkedBlockingQueue<String>, fragment: String) = assertTrue("Missing $fragment", logs.any { it.contains(fragment) })
+
+    @Test fun readFailureClosesCaptureRejectsSubscribeAndDetachedPollIsEmpty() {
+        val device = FaultyDevice(); val failed = CountDownLatch(1)
+        val capture = MicrophoneCapture(device, { _, text -> if (text.startsWith("capture reader failed")) failed.countDown() })
+        val sub = capture.subscribe("radio")
+        try {
+            device.input.offer(IllegalStateException("read failed"))
+            assertTrue(failed.await(3, TimeUnit.SECONDS)); assertTrue(capture.closed)
+            assertThrows(IllegalStateException::class.java) { sub.poll() }
+            assertThrows(IllegalStateException::class.java) { capture.subscribe("new") }
+        } finally { sub.close() }
+        assertNull(sub.poll()); assertEquals(1, device.closes.get())
+    }
+
+    @Test fun changedAndReleasedObserverFailuresDoNotLeakLastDevice() {
+        val device = FaultyDevice(); val logs = LinkedBlockingQueue<String>(); val releases = AtomicInteger()
+        val capture = MicrophoneCapture(device, { _, text -> logs.add(text) },
+            released = { releases.incrementAndGet(); error("release observer failed") }, changed = { error("change observer failed") })
+        val sub = capture.subscribe("radio")
+        repeat(3) { sub.close() }
+        assertEquals(1, device.stops.get()); assertEquals(1, device.closes.get()); assertEquals(1, releases.get())
+        diagnostic(logs, "changed observer failed"); diagnostic(logs, "released observer failed")
+    }
+
+    @Test fun interruptedCallerFinishesCleanupAndRetainsInterruptFlag() {
+        val device = FaultyDevice(); val releases = AtomicInteger()
+        val capture = MicrophoneCapture(device, { _, _ -> }, released = { releases.incrementAndGet() })
+        val sub = capture.subscribe("radio")
+        assertTrue(device.entered.await(3, TimeUnit.SECONDS))
+        val result = FutureTask {
+            Thread.currentThread().interrupt()
+            sub.close()
+            Thread.currentThread().isInterrupted
+        }
+        Thread(result, "interrupted-capture-owner").apply { isDaemon = true; start() }
+        assertTrue(result.get(3, TimeUnit.SECONDS))
+        assertEquals(1, device.closes.get()); assertEquals(1, releases.get())
+    }
+
+    @Test fun stopAndCloseFailureStillJoinReaderAndNotifyReleaseOnce() {
+        val device = FaultyDevice().apply { failStop = true; failClose = true }
+        val logs = LinkedBlockingQueue<String>(); val releases = AtomicInteger()
+        val capture = MicrophoneCapture(device, { _, text -> logs.add(text) }, released = { releases.incrementAndGet() })
+        val sub = capture.subscribe("radio")
+        assertTrue(device.entered.await(3, TimeUnit.SECONDS))
+        sub.close()
+        assertEquals(1, device.closes.get()); assertEquals(1, releases.get())
+        diagnostic(logs, "capture stop failed"); diagnostic(logs, "capture release failed")
+    }
+
+    @Test fun workerStartupFailureRollsBackWithoutReturningOrphanSubscription() {
+        for (factoryFailure in listOf(false, true)) {
+            val device = FaultyDevice(); val releases = AtomicInteger()
+            val capture = MicrophoneCapture(device, { _, _ -> }, released = { releases.incrementAndGet() }, threadFactory = { task ->
+                if (factoryFailure) error("thread factory denied")
+                object : Thread(task) { override fun start() { error("thread start denied") } }
+            })
+            assertThrows(IllegalStateException::class.java) { capture.subscribe("radio") }
+            assertTrue(capture.closed); assertTrue(capture.names.isEmpty())
+            assertEquals(1, device.closes.get()); assertEquals(1, releases.get())
+        }
+    }
+
+    @Test fun alreadyRunningFactoryResultCannotJoinOrInterruptCaller() {
+        val device = FaultyDevice()
+        val capture = MicrophoneCapture(device, { _, _ -> }, threadFactory = { Thread.currentThread() })
+        assertThrows(IllegalStateException::class.java) { capture.subscribe("radio") }
+        assertFalse(Thread.currentThread().isInterrupted)
+        assertEquals(1, device.closes.get())
+    }
+
+    @Test fun readerCannotCloseAndJoinItself() {
+        val device = FaultyDevice(); val failed = CountDownLatch(1); val logs = LinkedBlockingQueue<String>()
+        val capture = MicrophoneCapture(device, { _, text -> logs.add(text); if (text.startsWith("capture reader failed")) failed.countDown() })
+        val sub = capture.subscribe("radio")
+        device.onRead = { sub.close() }
+        try {
+            device.input.offer(MicrophoneCapture.Frame(shortArrayOf(1, 2), 1, 0))
+            assertTrue(failed.await(3, TimeUnit.SECONDS))
+            assertTrue(capture.closed)
+            diagnostic(logs, "close from reader rejected")
+        } finally { device.onRead = {}; sub.close() }
+        assertEquals(1, device.closes.get())
+    }
+
+    @Test fun controlCapacityReportsOverflowAndClosedPttIsRejected() {
+        val device = FaultyDevice(); val logs = LinkedBlockingQueue<String>()
+        val capture = MicrophoneCapture(device, { _, text -> logs.add(text) })
+        val sub = capture.subscribe("radio")
+        try {
+            repeat(32) { assertTrue(sub.ptt(it % 2 == 0)) }
+            assertFalse(sub.ptt(true)); diagnostic(logs, "control queue full")
+        } finally { sub.close() }
+        assertFalse(sub.ptt(true)); assertNull(sub.poll())
+    }
+
+    @Test fun observersExecuteOutsideCaptureMonitor() {
+        val device = FaultyDevice(); val locks = LinkedBlockingQueue<Boolean>()
+        lateinit var capture: MicrophoneCapture
+        capture = MicrophoneCapture(device, { _, _ -> }, released = { locks.add(Thread.holdsLock(capture)) },
+            changed = { locks.add(Thread.holdsLock(capture)) })
+        capture.subscribe("radio").close()
+        assertEquals(3, locks.size); assertTrue(locks.none { it })
+    }
+
+    @Test fun concurrentLastClosesReleaseDeviceExactlyOnce() {
+        repeat(32) {
+            val device = FaultyDevice(); val releases = AtomicInteger()
+            val capture = MicrophoneCapture(device, { _, _ -> }, released = { releases.incrementAndGet() })
+            val one = capture.subscribe("radio"); val two = capture.subscribe("stream")
+            val go = CountDownLatch(1)
+            val workers = listOf(one, two).map { sub -> FutureTask { go.await(); sub.close() }.also { task ->
+                Thread(task).apply { isDaemon = true; start() }
+            } }
+            go.countDown(); workers.forEach { it.get(3, TimeUnit.SECONDS) }
+            assertEquals(1, device.closes.get()); assertEquals(1, releases.get())
+        }
+    }
 }
@@ -0,0 +1,14 @@
+plugins { id("org.jetbrains.kotlin.jvm") }
+
+kotlin {
+    jvmToolchain(17)
+    sourceSets.named("main") {
+        // Настоящие reader/очереди, независимый от router-tests Android API без аппаратных ресурсов.
+        kotlin.srcDir("../app/src/main/java")
+        kotlin.include("android/**", "com/utun/chat/data/AndroidMicrophoneCapture.kt",
+            "com/utun/chat/data/MicrophoneCapture.kt", "com/utun/chat/data/CaptureHostLog.kt")
+    }
+}
+
+dependencies { testImplementation("junit:junit:4.13.2") }
+tasks.test { testLogging { showStandardStreams = true } }
@@ -0,0 +1,114 @@
+package android.media
+
+import java.util.concurrent.CopyOnWriteArrayList
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.LinkedBlockingQueue
+import java.util.concurrent.atomic.AtomicInteger
+
+class AudioDeviceInfo(val id: Int, val type: Int = 15, val productName: String = "microphone-$id")
+object AudioFormat { const val CHANNEL_IN_MONO = 16; const val CHANNEL_IN_STEREO = 12; const val ENCODING_PCM_16BIT = 2 }
+object MediaRecorder {
+    object AudioSource { const val MIC = 1; const val VOICE_COMMUNICATION = 7; const val UNPROCESSED = 9; const val CAMCORDER = 5 }
+}
+class AudioTimestamp {
+    companion object { const val TIMEBASE_MONOTONIC = 0 }
+    var nanoTime = 0L
+    var framePosition = 0L
+}
+class AudioRecordingConfiguration(val isClientSilenced: Boolean)
+
+/** Android API model: NON_BLOCKING действительно возвращается без данных, timestamps/route/ошибки задаёт сценарий. */
+class AudioRecord(val source: Int, val rate: Int, val mask: Int, encoding: Int, val bufferBytes: Int) {
+    data class Chunk(val count: Int? = null, val value: Short = 12, val device: Int? = 7, val silenced: Boolean = false,
+        val timeNs: Long? = null, val position: Long? = null, val measured: Boolean = true, val onRead: () -> Unit = {})
+    class Plan {
+        val chunks = LinkedBlockingQueue<Chunk>()
+        @Volatile var failures = emptySet<String>()
+        var initialized = true
+        var channels: Int? = null
+        var failChannelsAt = 0
+        val channelQueries = AtomicInteger()
+        var acceptInput = true
+        var starts = true
+        var initialDevice: Int? = 7
+        var clock: () -> Long = System::nanoTime
+        var holdRead: CountDownLatch? = null
+        val readEntered = CountDownLatch(1)
+        val stopEntered = CountDownLatch(1)
+        val readCalls = AtomicInteger()
+        val stopCalls = AtomicInteger()
+        val releaseCalls = AtomicInteger()
+        val modes = CopyOnWriteArrayList<Int>()
+        var released = false
+    }
+    companion object {
+        const val STATE_INITIALIZED = 1; const val RECORDSTATE_RECORDING = 3; const val RECORDSTATE_STOPPED = 1
+        const val READ_BLOCKING = 0; const val READ_NON_BLOCKING = 1; const val SUCCESS = 0
+        const val ERROR_INVALID_OPERATION = -3; const val ERROR_DEAD_OBJECT = -6
+        val plans = LinkedBlockingQueue<Plan>()
+        val created = CopyOnWriteArrayList<AudioRecord>()
+        val minimums = mutableMapOf<Int, Int>()
+        val masks = CopyOnWriteArrayList<Int>()
+        var constructorFailures = emptySet<Int>()
+        fun reset() { plans.clear(); created.clear(); minimums.clear(); masks.clear(); constructorFailures = emptySet() }
+        fun getMinBufferSize(rate: Int, mask: Int, encoding: Int): Int {
+            check(rate == 48000 && encoding == AudioFormat.ENCODING_PCM_16BIT)
+            return minimums[mask] ?: 960
+        }
+    }
+    val plan: Plan
+    init {
+        masks.add(mask)
+        check(mask !in constructorFailures) { "constructor failed mask=$mask" }
+        plan = requireNotNull(plans.poll()) { "missing record plan" }
+        created.add(this)
+        check(encoding == AudioFormat.ENCODING_PCM_16BIT && rate == 48000)
+    }
+    private var totalSamples = 0L
+    private var last: Chunk? = null
+    private var actual: Int? = plan.initialDevice
+    private var recording = RECORDSTATE_STOPPED
+    private fun operation(name: String) { check(name !in plan.failures) { "record operation failed: $name" } }
+    val state: Int get() { operation("state"); return if (plan.initialized) STATE_INITIALIZED else 0 }
+    val channelCount: Int get() {
+        check(plan.channelQueries.incrementAndGet() != plan.failChannelsAt) { "record channel query failed" }
+        operation("channels"); return plan.channels ?: if (mask == AudioFormat.CHANNEL_IN_STEREO) 2 else 1
+    }
+    val recordingState: Int get() { operation("recording"); return recording }
+    val routedDevice: AudioDeviceInfo? get() { operation("route"); return actual?.let { AudioDeviceInfo(it) } }
+    val activeRecordingConfiguration: AudioRecordingConfiguration get() {
+        operation("configuration"); return AudioRecordingConfiguration(last?.silenced == true)
+    }
+    fun setPreferredDevice(input: AudioDeviceInfo): Boolean { operation("prefer"); return plan.acceptInput }
+    fun startRecording() { operation("start"); if (plan.starts) recording = RECORDSTATE_RECORDING; operation("start after") }
+    fun read(pcm: ShortArray, offset: Int, size: Int, mode: Int): Int {
+        plan.readCalls.incrementAndGet(); plan.modes.add(mode); plan.readEntered.countDown()
+        operation("read")
+        plan.holdRead?.await()
+        val chunk = if (mode == READ_BLOCKING) plan.chunks.take() else plan.chunks.poll() ?: return 0
+        chunk.onRead()
+        last = chunk; actual = chunk.device
+        val count = chunk.count ?: size
+        if (count > 0) {
+            repeat(minOf(count, size)) { pcm[offset + it] = chunk.value }
+            totalSamples += count
+        }
+        return count
+    }
+    fun getTimestamp(stamp: AudioTimestamp, timebase: Int): Int {
+        check(timebase == AudioTimestamp.TIMEBASE_MONOTONIC); operation("timestamp")
+        val chunk = requireNotNull(last)
+        if (!chunk.measured) return ERROR_INVALID_OPERATION
+        stamp.nanoTime = chunk.timeNs ?: plan.clock()
+        stamp.framePosition = chunk.position ?: (totalSamples / channelCount - 960).coerceAtLeast(0)
+        return SUCCESS
+    }
+    fun stop() {
+        plan.stopCalls.incrementAndGet(); plan.stopEntered.countDown(); operation("stop")
+        recording = RECORDSTATE_STOPPED
+        plan.chunks.offer(Chunk(count = ERROR_INVALID_OPERATION))
+    }
+    fun release() {
+        plan.releaseCalls.incrementAndGet(); operation("release"); plan.released = true; recording = RECORDSTATE_STOPPED
+    }
+}
@@ -0,0 +1,12 @@
+package android.os
+
+/** SDK/priority seam отдельного hardware-reader JVM процесса. */
+object Build { object VERSION { var SDK_INT = 36 } }
+object Process {
+    const val THREAD_PRIORITY_AUDIO = -16
+    var failPriority = false
+    fun setThreadPriority(priority: Int) {
+        check(priority == THREAD_PRIORITY_AUDIO)
+        if (failPriority) throw SecurityException("audio priority denied")
+    }
+}
@@ -0,0 +1,13 @@
+package com.utun.chat.data
+
+import java.util.concurrent.LinkedBlockingQueue
+
+/** Структурированные логи настоящего reader; управление ошибками не заменяется fixture. */
+object LogManager {
+    data class Entry(val level: String, val category: String, val message: String)
+    val entries = LinkedBlockingQueue<Entry>()
+    fun addLog(level: String, category: String, message: String) {
+        entries.add(Entry(level, category, message))
+        System.err.println("[$level] [$category] $message")
+    }
+}
@@ -0,0 +1,291 @@
+package com.utun.chat.data
+
+import android.media.AudioDeviceInfo
+import android.media.AudioFormat
+import android.media.AudioRecord
+import android.media.MediaRecorder
+import android.os.Build
+import android.os.Process
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.FutureTask
+import java.util.concurrent.LinkedBlockingQueue
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicInteger
+import java.util.concurrent.atomic.AtomicLong
+import org.junit.After
+import org.junit.Assert.*
+import org.junit.Before
+import org.junit.Test
+
+/** Настоящий AudioRecord reader: частичные чтения, измеренная шкала, routing и полный lifecycle без микрофона ОС. */
+class AndroidMicrophoneCaptureTest {
+    private val now = AtomicLong(1_000_000_000L)
+    private val input = AudioDeviceInfo(7)
+    private val owners = mutableListOf<AndroidMicrophoneCapture>()
+    private val frames = LinkedBlockingQueue<MicrophoneCapture.Frame>()
+    private val errors = LinkedBlockingQueue<Exception>()
+    private val confirmations = AtomicInteger()
+    private val confirmed = CountDownLatch(1)
+
+    @Before fun prepare() {
+        AudioRecord.reset(); Process.failPriority = false; Build.VERSION.SDK_INT = 36; LogManager.entries.clear()
+    }
+
+    @After fun release() {
+        AudioRecord.created.forEach { it.plan.failures = emptySet(); it.plan.holdRead?.countDown() }
+        owners.forEach { it.close() }
+        // Частично созданный объект остаётся в fixture при инъекции ошибки release самого Android API.
+        AudioRecord.created.filter { !it.plan.released }.forEach { it.release() }
+        assertTrue(AudioRecord.created.all { it.plan.released })
+    }
+
+    private fun plan() = AudioRecord.Plan().also { it.clock = now::get; AudioRecord.plans.add(it) }
+    private fun open(stereo: Boolean = false, voice: Boolean = false, source: Int = 0,
+        accept: (MicrophoneCapture.Frame) -> Boolean = { frames.add(it); true },
+        failed: (Exception) -> Unit = { errors.add(it) },
+        confirmation: () -> Unit = { confirmations.incrementAndGet(); confirmed.countDown() }): AndroidMicrophoneCapture =
+        AndroidMicrophoneCapture.open(input, voice, stereo, source, 44L, accept, failed, confirmation, now::get).also { owners.add(it) }
+    private fun frame(): MicrophoneCapture.Frame = requireNotNull(frames.poll(3, TimeUnit.SECONDS)) { "No accepted PCM: ${LogManager.entries}" }
+    private fun failure(fragment: String): Exception = requireNotNull(errors.poll(3, TimeUnit.SECONDS)).also {
+        assertTrue("Expected $fragment: $it", it.message?.contains(fragment) == true)
+    }
+    private fun log(fragment: String) = assertTrue("Missing diagnostic $fragment", LogManager.entries.any { it.message.contains(fragment) })
+
+    @Test fun monoPcmIsStereoAndPreservesGenerationAndSequence() {
+        val plan = plan(); val capture = open()
+        plan.chunks.add(AudioRecord.Chunk(value = 31))
+        val first = frame()
+        assertEquals(1920, first.pcm.size); assertTrue(first.pcm.all { it == 31.toShort() })
+        assertEquals(44L, first.generation); assertEquals(0L, first.sequence)
+        now.addAndGet(20_000_000L); plan.chunks.add(AudioRecord.Chunk(value = 42))
+        val second = frame(); assertEquals(960L, second.sequence); assertTrue(second.pcm.all { it == 42.toShort() })
+        capture.close(); assertEquals(1, confirmations.get())
+        assertTrue(plan.modes.all { it == AudioRecord.READ_NON_BLOCKING })
+    }
+
+    @Test fun partialStereoReadsProduceOneWholeFrame() {
+        val plan = plan(); open(stereo = true)
+        plan.chunks.add(AudioRecord.Chunk(count = 480, value = 1))
+        plan.chunks.add(AudioRecord.Chunk(count = 1440, value = 2))
+        val frame = frame()
+        assertTrue(frame.pcm.take(480).all { it == 1.toShort() })
+        assertTrue(frame.pcm.drop(480).all { it == 2.toShort() })
+        assertEquals(0L, frame.sequence)
+    }
+
+    @Test fun sourceAndChannelPolicyMatchVoiceAndStereoConfiguration() {
+        for ((setting, expected) in listOf(0 to MediaRecorder.AudioSource.VOICE_COMMUNICATION, 1 to MediaRecorder.AudioSource.MIC,
+            2 to MediaRecorder.AudioSource.UNPROCESSED, 3 to MediaRecorder.AudioSource.CAMCORDER)) {
+            plan(); open(source = setting).close()
+            assertEquals(expected, AudioRecord.created.last().source)
+        }
+        plan(); open(stereo = true).close()
+        assertEquals(MediaRecorder.AudioSource.CAMCORDER, AudioRecord.created.last().source)
+        plan(); open(stereo = true, voice = true).close()
+        assertEquals(AudioFormat.CHANNEL_IN_MONO, AudioRecord.created.last().mask)
+        assertEquals(MediaRecorder.AudioSource.VOICE_COMMUNICATION, AudioRecord.created.last().source)
+    }
+
+    @Test fun rejectedStereoCleansFirstRecordAndFallsBackToMono() {
+        val stereo = plan().apply { acceptInput = false }; val mono = plan()
+        open(stereo = true).close()
+        assertEquals(listOf(AudioFormat.CHANNEL_IN_STEREO, AudioFormat.CHANNEL_IN_MONO), AudioRecord.masks.toList())
+        assertEquals(1, stereo.releaseCalls.get()); assertEquals(1, mono.releaseCalls.get())
+    }
+
+    @Test fun unsupportedStereoSkipsConstructorAndUsesMono() {
+        AudioRecord.minimums[AudioFormat.CHANNEL_IN_STEREO] = -2
+        plan(); open(stereo = true).close()
+        assertEquals(listOf(AudioFormat.CHANNEL_IN_MONO), AudioRecord.masks.toList())
+    }
+
+    @Test fun openFailureMatrixReleasesAllocatedRecord() {
+        for (operation in listOf("state", "prefer", "start", "start after", "recording", "channels")) {
+            val plan = plan().apply { failures = setOf(operation) }
+            assertThrows(IllegalStateException::class.java) { open() }
+            assertEquals(operation, 1, plan.releaseCalls.get())
+            assertTrue(plan.released)
+        }
+    }
+
+    @Test fun uninitializedOrNotStartedRecordIsNotPublished() {
+        val uninitialized = plan().apply { initialized = false }
+        assertThrows(IllegalStateException::class.java) { open() }
+        assertTrue(uninitialized.released)
+        val stopped = plan().apply { starts = false }
+        assertThrows(IllegalStateException::class.java) { open() }
+        assertTrue(stopped.released)
+        assertTrue(owners.isEmpty())
+    }
+
+    @Test fun releaseFailureDuringOpenAbortsFallbackAndKeepsOriginalCause() {
+        val plan = plan().apply { acceptInput = false; failures = setOf("release") }
+        val error = assertThrows(IllegalStateException::class.java) { open(stereo = true) }
+        assertTrue(error.message!!.contains("fallback aborted"))
+        assertTrue(error.cause!!.message!!.contains("Input selection rejected"))
+        assertEquals(1, error.cause!!.suppressed.size)
+        assertEquals(1, AudioRecord.created.size)
+        assertEquals(1, plan.releaseCalls.get()); log("capture open cleanup failed")
+    }
+
+    @Test fun setupFailureInsideWorkerReachesFailureCallback() {
+        plan().failChannelsAt = 3
+        open()
+        failure("channel query failed")
+    }
+
+    @Test fun deniedPriorityStillProducesUsefulPcm() {
+        Process.failPriority = true
+        val plan = plan(); open(); plan.chunks.add(AudioRecord.Chunk())
+        frame(); log("capture audio priority unavailable"); assertTrue(errors.isEmpty())
+    }
+
+    @Test fun noDataBeforeConfirmationFailsWithoutBlockingRead() {
+        val plan = plan(); open()
+        assertTrue(plan.readEntered.await(3, TimeUnit.SECONDS))
+        now.addAndGet(6_000_000_000L)
+        failure("data timeout")
+        assertTrue(plan.modes.all { it == AudioRecord.READ_NON_BLOCKING })
+    }
+
+    @Test fun stalledDataAfterConfirmationUsesOwnRecoveryDeadline() {
+        val plan = plan(); open(); plan.chunks.add(AudioRecord.Chunk())
+        frame(); assertTrue(confirmed.await(3, TimeUnit.SECONDS))
+        now.addAndGet(1_100_000_000L)
+        failure("data timeout")
+    }
+
+    @Test fun nativeReadErrorsAreReportedAndDoNotPublishPartialPcm() {
+        for (code in listOf(-2, -3, -6)) {
+            val plan = plan(); val capture = open()
+            plan.chunks.add(AudioRecord.Chunk(count = 240))
+            plan.chunks.add(AudioRecord.Chunk(count = code))
+            failure("read=$code"); assertTrue(frames.isEmpty()); capture.close()
+        }
+    }
+
+    @Test fun malformedStereoCountFailsWithContext() {
+        val plan = plan(); open(stereo = true)
+        plan.chunks.add(AudioRecord.Chunk(count = 1))
+        failure("channels=2")
+    }
+
+    @Test fun startupRouteOrSilencingMismatchCannotPublishPcm() {
+        for ((device, silenced) in listOf(8 to false, 7 to true)) {
+            val plan = plan(); val capture = open()
+            plan.chunks.add(AudioRecord.Chunk(device = device, silenced = silenced))
+            assertTrue(plan.readEntered.await(3, TimeUnit.SECONDS))
+            // Следующая точка clock не истекает до того, как metadata mismatch действительно проверен.
+            awaitLog("capture waiting metadata")
+            assertTrue(frames.isEmpty())
+            now.addAndGet(6_000_000_000L); failure("timeout")
+            capture.close(); LogManager.entries.clear()
+        }
+    }
+
+    @Test fun confirmedRouteLossImmediatelyReportsFailure() {
+        val plan = plan(); open(); plan.chunks.add(AudioRecord.Chunk()); frame()
+        plan.chunks.add(AudioRecord.Chunk(device = 8))
+        failure("route lost")
+    }
+
+    @Test fun missingTimestampAfterReadyGetsRecoveryBudgetAndCanResume() {
+        val plan = plan(); open(); plan.chunks.add(AudioRecord.Chunk()); frame()
+        now.addAndGet(100_000_000L); plan.chunks.add(AudioRecord.Chunk(measured = false))
+        awaitLog("capture waiting metadata")
+        now.addAndGet(20_000_000L); plan.chunks.add(AudioRecord.Chunk())
+        val next = frame(); assertEquals(1920L, next.sequence)
+        assertTrue(errors.isEmpty()); assertEquals(1, confirmations.get())
+    }
+
+    @Test fun timestampRecoveryBudgetExpiresAndIsDiagnosed() {
+        val plan = plan(); open(); plan.chunks.add(AudioRecord.Chunk()); frame()
+        now.addAndGet(100_000_000L); plan.chunks.add(AudioRecord.Chunk(measured = false))
+        awaitLog("capture waiting metadata")
+        plan.chunks.add(AudioRecord.Chunk(measured = false, onRead = { now.addAndGet(1_100_000_000L) }))
+        failure("confirmation timeout")
+    }
+
+    @Test fun futureOrNegativeTimestampCannotConfirmCapture() {
+        for (time in listOf(0L, -1L, now.get() + 100_000_000L)) {
+            val plan = plan(); val capture = open()
+            plan.chunks.add(AudioRecord.Chunk(timeNs = time))
+            awaitLog("capture waiting metadata")
+            assertTrue(frames.isEmpty()); assertEquals(0, confirmations.get())
+            capture.close(); LogManager.entries.clear()
+        }
+    }
+
+    @Test fun prolongedBackpressureFailsWithoutFalseConfirmation() {
+        val plan = plan(); open(accept = { false })
+        plan.chunks.add(AudioRecord.Chunk())
+        awaitReads(plan, 2)
+        plan.chunks.add(AudioRecord.Chunk(onRead = { now.addAndGet(1_100_000_000L) }))
+        failure("exceeds capture reserve"); assertEquals(0, confirmations.get())
+    }
+
+    @Test fun closeIsIdempotentEvenWhenStopFails() {
+        val plan = plan().apply { failures = setOf("stop") }
+        val capture = open(); assertTrue(plan.readEntered.await(3, TimeUnit.SECONDS))
+        repeat(3) { capture.close() }
+        assertEquals(1, plan.stopCalls.get()); assertEquals(1, plan.releaseCalls.get())
+        assertTrue(errors.isEmpty()); log("capture stop input=7")
+    }
+
+    @Test fun closePreservesInterruptedCallerAndReleaseCanRetry() {
+        val plan = plan().apply { failures = setOf("release") }; val capture = open()
+        assertThrows(IllegalStateException::class.java) { capture.close() }
+        assertEquals(1, plan.releaseCalls.get())
+        plan.failures = emptySet()
+        val result = FutureTask {
+            Thread.currentThread().interrupt(); capture.close(); Thread.currentThread().isInterrupted
+        }
+        Thread(result).apply { isDaemon = true; start() }
+        assertTrue(result.get(3, TimeUnit.SECONDS))
+        assertEquals(1, plan.stopCalls.get()); assertEquals(2, plan.releaseCalls.get())
+    }
+
+    @Test fun interruptedJoinWaitsForReaderBeforeRelease() {
+        val gate = CountDownLatch(1); val plan = plan().apply { holdRead = gate }
+        val capture = open(); assertTrue(plan.readEntered.await(3, TimeUnit.SECONDS))
+        val done = FutureTask { capture.close(); Thread.currentThread().isInterrupted }
+        val closer = Thread(done).apply { isDaemon = true; start() }
+        try {
+            assertTrue(plan.stopEntered.await(3, TimeUnit.SECONDS)); closer.interrupt()
+            awaitLog("capture join interrupted")
+            assertEquals(0, plan.releaseCalls.get())
+        } finally { gate.countDown() }
+        assertTrue(done.get(3, TimeUnit.SECONDS)); assertEquals(1, plan.releaseCalls.get())
+    }
+
+    @Test fun concurrentCloseReleasesExactlyOnce() {
+        val plan = plan(); val capture = open(); val go = CountDownLatch(1)
+        val calls = List(8) { FutureTask { go.await(); capture.close() }.also { task ->
+            Thread(task).apply { isDaemon = true; start() }
+        } }
+        go.countDown(); calls.forEach { it.get(3, TimeUnit.SECONDS) }
+        assertEquals(1, plan.stopCalls.get()); assertEquals(1, plan.releaseCalls.get())
+    }
+
+    @Test fun selfCloseInCallbackIsRejectedAndFailureCallbackExceptionIsLogged() {
+        val plan = plan(); lateinit var capture: AndroidMicrophoneCapture
+        capture = open(accept = { capture.close(); true }, failed = { errors.add(it); error("failure observer denied") })
+        plan.chunks.add(AudioRecord.Chunk())
+        failure("own reader"); capture.close()
+        log("capture close from reader rejected"); log("capture failure callback threw")
+    }
+
+    private fun awaitLog(fragment: String) {
+        val deadline = System.nanoTime() + 3_000_000_000L
+        while (System.nanoTime() < deadline) {
+            if (LogManager.entries.any { it.message.contains(fragment) }) return
+            java.util.concurrent.locks.LockSupport.parkNanos(1_000_000L)
+        }
+        error("Missing diagnostic $fragment: ${LogManager.entries}")
+    }
+    private fun awaitReads(plan: AudioRecord.Plan, count: Int) {
+        val deadline = System.nanoTime() + 3_000_000_000L
+        while (plan.readCalls.get() < count && System.nanoTime() < deadline) java.util.concurrent.locks.LockSupport.parkNanos(1_000_000L)
+        assertTrue(plan.readCalls.get() >= count)
+    }
+}
@@ -18,3 +18,4 @@ rootProject.name = "chatgui-android"
 include(":app")
 include(":audio-probe")
 include(":router-tests")
+include(":capture-tests")