feat(reaction): Add Conversational Awareness media reaction

Reacts to the AirPods speaking-detection event (AAP 0x4B): lowers media volume by a configurable amount or pauses playback when you start talking, and reverts when you stop. Per-profile, Pro-gated, opt-in (default off). Decodes the speaking status from the last payload byte ({1,2}=start, {6,8,9}=stop, else keep-alive); engage/disengage with a frame-idle stale timeout to recover a dropped stop, plus disconnect and service-stop cleanup.
2026-09-15 10:46:12 -04:00 · 2026-05-25 23:55:55 +02:00
parent 78790f73df
commit dbbfa75429
20 changed files with 1010 additions and 12 deletions
@@ -148,6 +148,48 @@ class DefaultAapDeviceProfileTest : BaseAapSessionTest() {
        @Test fun `decode unknown value returns null`() { profile.decodeSetting(settingsMessage(0x28, 0x00)).shouldBeNull() }
    }

+    // ── Conversational Awareness State (push-only 0x4B) ──────
+
+    @Nested
+    inner class ConversationalAwarenessStateTests {
+        @Test fun `4-byte frame status 1 is speaking`() {
+            decodeSetting<AapSetting.ConversationalAwarenessState>("04 00 04 00 4B 00 02 00 01 01").let {
+                it.speaking shouldBe true
+                it.rawValue shouldBe 1
+            }
+        }
+
+        @Test fun `4-byte frame status 2 is speaking`() {
+            decodeSetting<AapSetting.ConversationalAwarenessState>("04 00 04 00 4B 00 02 00 01 02").speaking shouldBe true
+        }
+
+        @Test fun `4-byte frame stop status 9 is not speaking`() {
+            decodeSetting<AapSetting.ConversationalAwarenessState>("04 00 04 00 4B 00 02 00 01 09").let {
+                it.speaking shouldBe false
+                it.rawValue shouldBe 9
+            }
+        }
+
+        @Test fun `4-byte frame intermediate status 4 is not speaking`() {
+            decodeSetting<AapSetting.ConversationalAwarenessState>("04 00 04 00 4B 00 02 00 01 04").speaking shouldBe false
+        }
+
+        @Test fun `legacy single-byte frame status 0 is not speaking`() {
+            decodeSetting<AapSetting.ConversationalAwarenessState>("04 00 04 00 4B 00 00").let {
+                it.speaking shouldBe false
+                it.rawValue shouldBe 0
+            }
+        }
+
+        @Test fun `truncated 3-byte payload returns null`() {
+            profile.decodeSetting(aapMessage("04 00 04 00 4B 00 02 00 01")).shouldBeNull()
+        }
+
+        @Test fun `invalid 4-byte prefix returns null`() {
+            profile.decodeSetting(aapMessage("04 00 04 00 4B 00 02 00 02 01")).shouldBeNull()
+        }
+    }
+
    // ── Press Speed ──────────────────────────────────────────

    @Nested
@@ -8,6 +8,7 @@ import eu.darken.capod.pods.core.apple.aap.protocol.AapMessage
 import eu.darken.capod.pods.core.apple.aap.protocol.AapPacket
 import eu.darken.capod.pods.core.apple.aap.protocol.AapSetting
 import eu.darken.capod.pods.core.apple.aap.protocol.AapSleepEvent
+import eu.darken.capod.pods.core.apple.aap.protocol.ConversationAwarenessEvent
 import eu.darken.capod.pods.core.apple.aap.protocol.StemPressEvent
 import io.kotest.matchers.collections.shouldBeEmpty
 import io.kotest.matchers.nulls.shouldBeNull
@@ -497,6 +498,50 @@ class AapSessionEngineTest : BaseTest() {
        }
    }

+    // ── Conversational Awareness ─────────────────────────────
+
+    @Nested
+    inner class ConversationalAwarenessTests {
+
+        private fun caProfile(status: Int) = mockProfile {
+            every { decodeSetting(any()) } returns settingPair(
+                AapSetting.ConversationalAwarenessState(
+                    speaking = status in ConversationAwarenessEvent.SPEAKING_STATUSES,
+                    rawValue = status,
+                ),
+            )
+        }
+
+        private suspend fun TestScope.firstEventFor(status: Int): ConversationAwarenessEvent {
+            val engine = AapSessionEngine(caProfile(status), timeSource)
+            engine.start(this)
+            var emitted: ConversationAwarenessEvent? = null
+            val job = launch { emitted = engine.conversationalAwarenessEvents.first() }
+            engine.processMessage(dummyMessage(commandType = 0x004B))
+            job.join()
+            return emitted!!
+        }
+
+        @Test
+        fun `status 1 and 2 emit START`() = runTest(UnconfinedTestDispatcher()) {
+            firstEventFor(1) shouldBe ConversationAwarenessEvent.START
+            firstEventFor(2) shouldBe ConversationAwarenessEvent.START
+        }
+
+        @Test
+        fun `status 6, 8, 9 emit STOP`() = runTest(UnconfinedTestDispatcher()) {
+            firstEventFor(6) shouldBe ConversationAwarenessEvent.STOP
+            firstEventFor(8) shouldBe ConversationAwarenessEvent.STOP
+            firstEventFor(9) shouldBe ConversationAwarenessEvent.STOP
+        }
+
+        @Test
+        fun `intermediate status emits HOLD (keep-alive)`() = runTest(UnconfinedTestDispatcher()) {
+            firstEventFor(3) shouldBe ConversationAwarenessEvent.HOLD
+            firstEventFor(0x0B) shouldBe ConversationAwarenessEvent.HOLD
+        }
+    }
+
    // ── Inference ───────────────────────────────────────────

    @Nested
@@ -0,0 +1,275 @@
+package eu.darken.capod.reaction.core.conversation
+
+import eu.darken.capod.common.MediaControl
+import eu.darken.capod.common.bluetooth.BluetoothAddress
+import eu.darken.capod.monitor.core.DeviceMonitor
+import eu.darken.capod.monitor.core.PodDevice
+import eu.darken.capod.pods.core.apple.aap.AapConnectionManager
+import eu.darken.capod.pods.core.apple.aap.AapPodState
+import eu.darken.capod.pods.core.apple.aap.protocol.ConversationAwarenessEvent
+import eu.darken.capod.profiles.core.ReactionConfig
+import io.mockk.coEvery
+import io.mockk.coVerify
+import io.mockk.every
+import io.mockk.mockk
+import io.mockk.verify
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
+import kotlinx.coroutines.test.advanceTimeBy
+import kotlinx.coroutines.test.runCurrent
+import kotlinx.coroutines.test.runTest
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+import testhelpers.BaseTest
+import testhelpers.TestTimeSource
+
+class ConversationReactionTest : BaseTest() {
+
+    private val primaryAddress: BluetoothAddress = "AA:BB:CC:DD:EE:FF"
+    private val otherAddress: BluetoothAddress = "11:22:33:44:55:66"
+
+    // Mirror of ConversationReaction.STALE_TIMEOUT_MS (private there).
+    private val staleTimeoutMs = 12_000L
+
+    private lateinit var eventsFlow: MutableSharedFlow<Pair<BluetoothAddress, ConversationAwarenessEvent>>
+    private lateinit var statesFlow: MutableStateFlow<Map<BluetoothAddress, AapPodState>>
+    private lateinit var devicesFlow: MutableStateFlow<List<PodDevice>>
+    private lateinit var aapManager: AapConnectionManager
+    private lateinit var deviceMonitor: DeviceMonitor
+    private lateinit var mediaControl: MediaControl
+    private lateinit var timeSource: TestTimeSource
+
+    private fun mockPodDevice(
+        address: BluetoothAddress,
+        action: ConversationAction,
+        reduction: Int = 50,
+        worn: Boolean = true,
+    ): PodDevice = mockk(relaxed = true) {
+        every { profileId } returns address
+        every { this@mockk.address } returns address
+        every { reactions } returns ReactionConfig(
+            conversationAction = action,
+            conversationVolumeReduction = reduction,
+        )
+        every { isBeingWorn } returns worn
+    }
+
+    @BeforeEach
+    fun setup() {
+        eventsFlow = MutableSharedFlow(extraBufferCapacity = 16)
+        statesFlow = MutableStateFlow(mapOf(primaryAddress to mockk(relaxed = true)))
+        devicesFlow = MutableStateFlow(listOf(mockPodDevice(primaryAddress, ConversationAction.LOWER_VOLUME)))
+        aapManager = mockk(relaxed = true) {
+            every { conversationalAwarenessEvents } returns eventsFlow
+            every { allStates } returns statesFlow
+        }
+        deviceMonitor = mockk(relaxed = true) {
+            every { devices } returns devicesFlow
+        }
+        mediaControl = mockk(relaxed = true) {
+            coEvery { sendPause(any()) } returns true
+            every { isPlaying } returns false
+            every { duckMusicVolume(any()) } returns MediaControl.VolumeDuck(priorVolume = 10, appliedVolume = 5)
+            every { currentMusicVolume() } returns 5
+        }
+        timeSource = TestTimeSource()
+    }
+
+    // appScope = the test scope so the stale timer runs on the controllable virtual clock.
+    private fun TestScope.launchReaction() = ConversationReaction(
+        aapManager = aapManager,
+        deviceMonitor = deviceMonitor,
+        mediaControl = mediaControl,
+        appScope = this,
+        timeSource = timeSource,
+    ).monitor().launchIn(this)
+
+    private suspend fun TestScope.emit(address: BluetoothAddress, event: ConversationAwarenessEvent) {
+        eventsFlow.emit(address to event)
+        runCurrent()
+    }
+
+    @Test
+    fun `LOWER_VOLUME start ducks, stop restores`() = runTest(UnconfinedTestDispatcher()) {
+        val job = launchReaction()
+
+        emit(primaryAddress, ConversationAwarenessEvent.START)
+        verify(exactly = 1) { mediaControl.duckMusicVolume(50) }
+        verify(exactly = 0) { mediaControl.restoreMusicVolume(any()) }
+
+        emit(primaryAddress, ConversationAwarenessEvent.STOP)
+        verify(exactly = 1) { mediaControl.restoreMusicVolume(10) }
+        job.cancel()
+    }
+
+    @Test
+    fun `LOWER_VOLUME restores unconditionally even if the volume was moved meanwhile`() = runTest(UnconfinedTestDispatcher()) {
+        // Something else (e.g. another volume-manager app) moved the volume during the duck, so the
+        // current reading no longer matches what we applied — we must still restore to the saved prior.
+        every { mediaControl.currentMusicVolume() } returns 7
+        val job = launchReaction()
+
+        emit(primaryAddress, ConversationAwarenessEvent.START)
+        emit(primaryAddress, ConversationAwarenessEvent.STOP)
+
+        verify(exactly = 1) { mediaControl.restoreMusicVolume(10) }
+        job.cancel()
+    }
+
+    @Test
+    fun `LOWER_VOLUME owner leaving the AAP state map restores volume`() = runTest(UnconfinedTestDispatcher()) {
+        val job = launchReaction()
+
+        emit(primaryAddress, ConversationAwarenessEvent.START)
+        statesFlow.value = emptyMap() // device disconnected before STOP arrived
+        runCurrent()
+
+        verify(exactly = 1) { mediaControl.restoreMusicVolume(10) }
+        job.cancel()
+    }
+
+    @Test
+    fun `LOWER_VOLUME missed STOP restores via stale timeout`() = runTest(UnconfinedTestDispatcher()) {
+        val job = launchReaction()
+
+        emit(primaryAddress, ConversationAwarenessEvent.START)
+        verify(exactly = 1) { mediaControl.duckMusicVolume(50) }
+        verify(exactly = 0) { mediaControl.restoreMusicVolume(any()) }
+
+        // No STOP arrives; frames cease. After the stale timeout, volume must be restored.
+        advanceTimeBy(staleTimeoutMs + 500)
+        runCurrent()
+
+        verify(exactly = 1) { mediaControl.restoreMusicVolume(10) }
+        job.cancel()
+    }
+
+    @Test
+    fun `HOLD keep-alive resets the stale timer`() = runTest(UnconfinedTestDispatcher()) {
+        val job = launchReaction()
+
+        emit(primaryAddress, ConversationAwarenessEvent.START)
+        advanceTimeBy(8_000)
+        runCurrent()
+        emit(primaryAddress, ConversationAwarenessEvent.HOLD) // resets the timer
+        advanceTimeBy(8_000) // 8s since the HOLD — still within the window
+        runCurrent()
+        verify(exactly = 0) { mediaControl.restoreMusicVolume(any()) }
+
+        advanceTimeBy(5_000) // now >12s since the last frame
+        runCurrent()
+        verify(exactly = 1) { mediaControl.restoreMusicVolume(10) }
+        job.cancel()
+    }
+
+    @Test
+    fun `HOLD without a prior START does not engage`() = runTest(UnconfinedTestDispatcher()) {
+        val job = launchReaction()
+
+        emit(primaryAddress, ConversationAwarenessEvent.HOLD)
+
+        verify(exactly = 0) { mediaControl.duckMusicVolume(any()) }
+        job.cancel()
+    }
+
+    @Test
+    fun `PAUSE start pauses, stop resumes when worn and idle`() = runTest(UnconfinedTestDispatcher()) {
+        devicesFlow.value = listOf(mockPodDevice(primaryAddress, ConversationAction.PAUSE))
+        val job = launchReaction()
+
+        emit(primaryAddress, ConversationAwarenessEvent.START)
+        coVerify(exactly = 1) { mediaControl.sendPause(false) }
+
+        emit(primaryAddress, ConversationAwarenessEvent.STOP)
+        coVerify(exactly = 1) { mediaControl.sendPlay() }
+        job.cancel()
+    }
+
+    @Test
+    fun `PAUSE stop does not resume when pods not worn`() = runTest(UnconfinedTestDispatcher()) {
+        devicesFlow.value = listOf(mockPodDevice(primaryAddress, ConversationAction.PAUSE, worn = false))
+        val job = launchReaction()
+
+        emit(primaryAddress, ConversationAwarenessEvent.START)
+        emit(primaryAddress, ConversationAwarenessEvent.STOP)
+
+        coVerify(exactly = 0) { mediaControl.sendPlay() }
+        job.cancel()
+    }
+
+    @Test
+    fun `PAUSE stop does not resume when something is already playing`() = runTest(UnconfinedTestDispatcher()) {
+        devicesFlow.value = listOf(mockPodDevice(primaryAddress, ConversationAction.PAUSE))
+        val job = launchReaction()
+
+        emit(primaryAddress, ConversationAwarenessEvent.START)
+        every { mediaControl.isPlaying } returns true // user/app restarted playback during the talk
+        emit(primaryAddress, ConversationAwarenessEvent.STOP)
+
+        coVerify(exactly = 0) { mediaControl.sendPlay() }
+        job.cancel()
+    }
+
+    @Test
+    fun `PAUSE resumes on stop even if the action was switched away mid-talk`() = runTest(UnconfinedTestDispatcher()) {
+        devicesFlow.value = listOf(mockPodDevice(primaryAddress, ConversationAction.PAUSE))
+        val job = launchReaction()
+
+        emit(primaryAddress, ConversationAwarenessEvent.START)
+        coVerify(exactly = 1) { mediaControl.sendPause(false) }
+
+        // User changes the action mid-conversation; we must still undo the pause WE caused.
+        devicesFlow.value = listOf(mockPodDevice(primaryAddress, ConversationAction.LOWER_VOLUME))
+        emit(primaryAddress, ConversationAwarenessEvent.STOP)
+
+        coVerify(exactly = 1) { mediaControl.sendPlay() }
+        job.cancel()
+    }
+
+    @Test
+    fun `NOTHING action ignores speaking`() = runTest(UnconfinedTestDispatcher()) {
+        devicesFlow.value = listOf(mockPodDevice(primaryAddress, ConversationAction.NOTHING))
+        val job = launchReaction()
+
+        emit(primaryAddress, ConversationAwarenessEvent.START)
+
+        verify(exactly = 0) { mediaControl.duckMusicVolume(any()) }
+        coVerify(exactly = 0) { mediaControl.sendPause(any()) }
+        job.cancel()
+    }
+
+    @Test
+    fun `non-primary device start is ignored`() = runTest(UnconfinedTestDispatcher()) {
+        val job = launchReaction()
+
+        emit(otherAddress, ConversationAwarenessEvent.START)
+
+        verify(exactly = 0) { mediaControl.duckMusicVolume(any()) }
+        job.cancel()
+    }
+
+    @Test
+    fun `stop without a prior start is a no-op`() = runTest(UnconfinedTestDispatcher()) {
+        val job = launchReaction()
+
+        emit(primaryAddress, ConversationAwarenessEvent.STOP)
+
+        verify(exactly = 0) { mediaControl.restoreMusicVolume(any()) }
+        coVerify(exactly = 0) { mediaControl.sendPlay() }
+        job.cancel()
+    }
+
+    @Test
+    fun `duplicate start does not duck twice`() = runTest(UnconfinedTestDispatcher()) {
+        val job = launchReaction()
+
+        emit(primaryAddress, ConversationAwarenessEvent.START)
+        emit(primaryAddress, ConversationAwarenessEvent.START) // status 1 then 2 both classify as START
+
+        verify(exactly = 1) { mediaControl.duckMusicVolume(any()) }
+        job.cancel()
+    }
+}