fix(reaction): Restore volume when conversation end frame never arrives

With one AirPod in the case the pod deterministically drops the
terminal 0x4B frame: the wind-down flurry ends on a transitional
status (1,2,3,0xB,4 then nothing), so the reaction never disengaged
and the volume stayed low until the 5-minute backstop (#608).
Reproduced on Pro 3 (fw 6589, 6503) and Pro 2 USB-C (fw 6814) —
all three share one firmware train; with both pods worn the terminal
always arrives.

The pod sends no frames during active speech, so any non-START frame
means the wind-down has begun and a terminal is imminent. A HOLD now
arms a short 6s fuse instead of the 5-minute backstop: if no terminal
(or fresh START) follows, the reaction disengages anyway. The fuse
must stay above ~5s — gaps up to 2.8s were observed between
consecutive wind-down frames. STARTs still arm the long backstop,
since the pod stays engaged and silent against ambient noise for
20-30s+ and a short timeout there resumes media mid-conversation.

Validated on hardware on both models: 7/7 single-pod conversations
restored exactly 6s after the last wind-down frame.

app/src/main/java/eu/darken/capod/pods/core/apple/aap/protocol/ConversationAwarenessEvent.kt

@@ -3,19 +3,23 @@ package eu.darken.capod.pods.core.apple.aap.protocol
 /**
  * Classified Conversational Awareness signal derived from the status byte of a `0x4B` frame.
  *
- * Status-byte mapping (from live AirPods Pro 3 captures + the librepods project):
+ * Status-byte mapping (from live AirPods Pro 3 + Pro 2 USB-C captures — both models share one
+ * firmware train and a byte-identical protocol — plus the librepods project):
  * - `1`, `2` → [START] (wearer started / is speaking → engage the reaction)
- * - `5`, `6`, `8`, `9` → [STOP] (wearer stopped → disengage). `5` is the terminal value on fw `…6861`,
- *   which winds down `3`→`5` and never reaches `6/8/9`; `6/8/9` are the terminal values on fw `…6503`.
- * - any other value (`0`, `3`, `4`, `0x0B`, … and anything unrecognised) → [HOLD]: a transitional or
- *   unknown frame. It must NOT disengage the reaction — only an explicit terminal [STOP] does that.
+ * - `5`, `6`, `8`, `9` → [STOP] (wearer stopped → disengage). All four confirmed live on Pro 2
+ *   USB-C fw `…6814`; which one terminates a given flurry varies with how speech ended.
+ * - any other value (`0`, `3`, `4`, `7`, `0x0B`, … and anything unrecognised) → [HOLD]: a
+ *   transitional wind-down frame (`7` was only discovered on fw `…6814` — the set is open-ended,
+ *   so unknown values are deliberately classified as HOLD rather than guessed at).
  *
- * Frame cadence is firmware-dependent and the pod does NOT reliably stream keep-alives while you
- * talk: fw `…6861` sent only an onset (`1`,`2`) then NO `0x4B` frames for 21s of continuous speech
- * (proven still-speaking — it held its own CA/ANC-transparency engaged the whole time), then the
- * wind-down `3`,`5`. So frame-silence must NOT be read as "speaking ended"; disengage is driven by
- * the explicit terminal [STOP] frame. [ConversationReaction]'s stale timeout is only a long backstop
- * for a fully-dropped terminal frame, not the normal disengage path.
+ * The pod sends NO frames during active speech — it stays engaged (and silent) for as long as it
+ * hears nearby voices, 20-30s+ observed. So frame-silence must NOT be read as "speaking ended".
+ * Conversely, any non-START frame means the wind-down has begun: a short flurry of transitional
+ * and terminal frames (e.g. `3,0xB,4,8,9` or `3,5,7,8,9`). With only ONE pod worn (other in
+ * case/disconnected) the terminal is deterministically dropped — the flurry ends on a transitional
+ * `4` (#608; reproduced on Pro 3 and Pro 2 alike) — so [ConversationReaction] treats a HOLD as
+ * "terminal imminent" and arms a short fuse, with a long stale backstop for a fully-dropped flurry.
+ * A flurry's trailing frames may arrive after its terminal; they are ignored once disengaged.
  */
 enum class ConversationAwarenessEvent {
     START,

app/src/main/java/eu/darken/capod/reaction/core/conversation/ConversationReaction.kt

@@ -30,8 +30,10 @@ import kotlinx.coroutines.sync.withLock
 import kotlinx.coroutines.withContext
 import javax.inject.Inject
 import javax.inject.Singleton
+import kotlin.time.Duration
 import kotlin.time.Duration.Companion.milliseconds
 import kotlin.time.Duration.Companion.minutes
+import kotlin.time.Duration.Companion.seconds
 
 /**
  * Reacts to Conversational Awareness speaking transitions (AAP `0x4B`) by either lowering media
@@ -39,13 +41,16 @@ import kotlin.time.Duration.Companion.minutes
  * speaking stops. On Android the pod firmware does not duck audio itself, so CAPod performs it.
  *
  * Disengage is driven by the pod's explicit end-of-speech frame ([ConversationAwarenessEvent.STOP]).
- * The pod does NOT reliably stream keep-alive frames while you talk — on some firmware it sends an
- * onset ([ConversationAwarenessEvent.START]) then nothing for many seconds while speech continues
- * (observed: CA held engaged 21s with zero `0x4B` frames). So frame-silence must NOT be read as
- * "speaking ended". [STALE_TIMEOUT] is only a long backstop for the rare case where every terminal
- * frame is lost while the link stays up; a link drop is handled by the owner-disconnect revert.
- * [ConversationAwarenessEvent.HOLD] frames (transitional/unknown statuses) keep the reaction engaged
- * and refresh the backstop.
+ * The pod sends NO frames during active speech — it stays engaged for as long as it hears nearby
+ * voices (observed: CA held engaged 21-32s with zero `0x4B` frames, indefinitely against ambient
+ * noise). So frame-silence must NOT be read as "speaking ended"; after a START only the long
+ * [STALE_TIMEOUT] backstop applies, and a link drop is handled by the owner-disconnect revert.
+ *
+ * Because the pod is silent while engaged, any non-START frame is evidence the wind-down has begun.
+ * The wind-down is a short flurry of transitional ([ConversationAwarenessEvent.HOLD]) and terminal
+ * frames — but the terminal is sometimes dropped entirely (fw `…6589` emitted `3,0xB,4` then nothing,
+ * stranding the volume low — #608). So a HOLD frame re-arms the timer with the short
+ * [WIND_DOWN_TIMEOUT] fuse: if no terminal follows, speech is treated as ended anyway.
  *
  * State is a single global slot (media volume / playback is system-wide, not per-device) guarded by
  * a [Mutex] — events, AAP-state-removal, the stale timer, and monitor completion all mutate it.
@@ -108,7 +113,8 @@ class ConversationReaction @Inject constructor(
             val current = active
             if (current != null && current.owner == address) {
                 // Duplicate START for the same speaker — don't re-act, just keep the session alive.
-                restartStaleTimer(current)
+                // A START also cancels a pending wind-down fuse: the wearer is speaking again.
+                armDisengageTimer(current, STALE_TIMEOUT)
                 log(TAG) { "START from $address — already active ($action), keep-alive" }
                 return
             }
@@ -121,7 +127,7 @@ class ConversationReaction @Inject constructor(
                     if (paused) {
                         val record = Active(nextId(), address, Kind.Paused, timeSource.elapsedRealtime())
                         active = record
-                        restartStaleTimer(record)
+                        armDisengageTimer(record, STALE_TIMEOUT)
                         log(TAG, INFO) { "START on $address → paused media" }
                     } else {
                         active = null
@@ -144,7 +150,7 @@ class ConversationReaction @Inject constructor(
                             timeSource.elapsedRealtime(),
                         )
                         active = record
-                        restartStaleTimer(record)
+                        armDisengageTimer(record, STALE_TIMEOUT)
                         log(TAG, INFO) { "START on $address → ducked volume ${duck.priorVolume}→${duck.appliedVolume}" }
                     } else {
                         active = null
@@ -157,11 +163,15 @@ class ConversationReaction @Inject constructor(
         }
     }
 
-    /** Intermediate keep-alive frame: the wearer is still speaking — refresh the stale timer. */
+    /**
+     * Transitional wind-down frame. The pod is silent during active speech, so this frame means the
+     * wind-down has begun and a terminal frame is imminent — but firmware sometimes drops it (#608).
+     * Arm the short fuse: if no terminal (or fresh START) follows, disengage anyway.
+     */
     private suspend fun onSpeakingHold(address: BluetoothAddress) = mutex.withLock {
         val current = active ?: return
         if (current.owner != address) return
-        restartStaleTimer(current)
+        armDisengageTimer(current, WIND_DOWN_TIMEOUT)
     }
 
     private suspend fun onSpeakingStop(address: BluetoothAddress) {
@@ -247,21 +257,21 @@ class ConversationReaction @Inject constructor(
     }
 
     /**
-     * Must be called under [mutex]. (Re)starts the backstop timer for [record], reset on every frame
-     * (START/HOLD). If no frame arrives for [STALE_TIMEOUT] the session is force-ended — a last
-     * resort for a fully-dropped terminal frame, not the normal disengage. Identity-checked so a
-     * late timer can't disengage a newer session.
+     * Must be called under [mutex]. (Re)arms the disengage timer for [record] — every frame picks
+     * the fuse matching its meaning: START → [STALE_TIMEOUT] (active speech, frames cease for its
+     * whole duration), HOLD → [WIND_DOWN_TIMEOUT] (wind-down begun, terminal imminent). On expiry
+     * the session is force-ended. Identity-checked so a late timer can't disengage a newer session.
      */
-    private fun restartStaleTimer(record: Active) {
+    private fun armDisengageTimer(record: Active, timeout: Duration) {
         staleJob?.cancel()
         staleJob = appScope.launch {
-            delay(STALE_TIMEOUT)
+            delay(timeout)
             val primary = deviceMonitor.primaryDevice().first()
             mutex.withLock {
                 if (active?.id == record.id) {
                     val current = active!!
                     clearActive()
-                    disengage(current, primary, "stale timeout (frames ceased)")
+                    disengage(current, primary, "no terminal frame within $timeout")
                 }
             }
         }
@@ -273,16 +283,26 @@ class ConversationReaction @Inject constructor(
         private val TAG = logTag("Reaction", "Conversation")
 
         /**
-         * Pure backstop: disengage if no `0x4B` frame arrives for this long while engaged. This is
-         * NOT the normal disengage path — the pod does not stream keep-alives during speech, so a
-         * short timeout would fire mid-conversation (the original 12s value did exactly that).
-         * Normal disengage is the explicit terminal [ConversationAwarenessEvent.STOP] frame; this
-         * only recovers a fully-dropped terminal frame while the link stays up. Kept longer than
-         * [PAUSE_RESUME_WINDOW] so a back-stopped pause is never auto-resumed (only a duck is
-         * restored on stale — a stranded low volume is the worse failure).
+         * Backstop fuse while engaged with no wind-down evidence yet (only START frames seen).
+         * Must stay LONG: the pod sends zero frames during active speech and stays engaged against
+         * ambient noise, so a short timeout here resumes media mid-conversation (the original 12s
+         * value did exactly that). Only recovers a session whose entire wind-down flurry was lost
+         * while the link stays up. Kept longer than [PAUSE_RESUME_WINDOW] so a back-stopped pause
+         * is never auto-resumed (only a duck is restored — a stranded low volume is the worse
+         * failure).
          */
         private val STALE_TIMEOUT = 5.minutes
 
+        /**
+         * Short fuse armed by a transitional [ConversationAwarenessEvent.HOLD] frame: the wind-down
+         * has begun, so a terminal frame should follow within seconds. With only one pod worn the
+         * pod deterministically drops the terminal (#608, reproduced on Pro 3 and Pro 2) — this
+         * fuse disengages instead of stranding the volume low for [STALE_TIMEOUT]. Must be ≥ ~5s:
+         * gaps up to 2.8s were observed between consecutive wind-down frames, and each HOLD re-arms
+         * this fuse. A fresh START re-arms the long fuse (speaking resumed).
+         */
+        private val WIND_DOWN_TIMEOUT = 6.seconds
+
         /** A pause older than this no longer auto-resumes — unexpected late playback is worse. */
         private val PAUSE_RESUME_WINDOW = 2.minutes
     }

app/src/test/java/eu/darken/capod/reaction/core/conversation/ConversationReactionTest.kt

@@ -31,9 +31,11 @@ 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 (private there) — a long backstop, not the
-    // normal disengage path (which is the explicit STOP frame).
+    // Mirrors of ConversationReaction's private timing constants. STALE_TIMEOUT: long backstop
+    // while only START frames were seen. WIND_DOWN_TIMEOUT: short fuse armed by a HOLD frame
+    // (wind-down begun, terminal imminent — but sometimes dropped, #608).
     private val staleTimeoutMs = 5L * 60 * 1000
+    private val windDownTimeoutMs = 6_000L
 
     private lateinit var eventsFlow: MutableSharedFlow<Pair<BluetoothAddress, ConversationAwarenessEvent>>
     private lateinit var statesFlow: MutableStateFlow<Map<BluetoothAddress, AapPodState>>
@@ -149,20 +151,79 @@ class ConversationReactionTest : BaseTest() {
     }
 
     @Test
-    fun `HOLD keep-alive resets the stale backstop`() = runTest(UnconfinedTestDispatcher()) {
+    fun `dropped terminal frame restores via the wind-down fuse`() = runTest(UnconfinedTestDispatcher()) {
+        // Regression for #608 (fw …6589): the wind-down flurry 1,2,3,0xB,4 arrives but the terminal
+        // 8/9 is dropped entirely. The HOLD frames prove the wind-down began, so the short fuse must
+        // restore the volume — not strand it low until the 5-minute backstop.
+        val job = launchReaction()
+
+        emit(primaryAddress, ConversationAwarenessEvent.START) // 1
+        emit(primaryAddress, ConversationAwarenessEvent.START) // 2
+        emit(primaryAddress, ConversationAwarenessEvent.HOLD) // 3
+        emit(primaryAddress, ConversationAwarenessEvent.HOLD) // 0x0B
+        emit(primaryAddress, ConversationAwarenessEvent.HOLD) // 4 — last frame ever, no terminal
+        verify(exactly = 0) { mediaControl.restoreMusicVolume(any()) }
+
+        advanceTimeBy(windDownTimeoutMs + 500)
+        runCurrent()
+        verify(exactly = 1) { mediaControl.restoreMusicVolume(10) }
+        job.cancel()
+    }
+
+    @Test
+    fun `HOLD frames re-arm the wind-down fuse so intra-flurry gaps do not fire it`() =
+        runTest(UnconfinedTestDispatcher()) {
+            // Gaps of up to 2.8s were observed between consecutive wind-down frames — each HOLD
+            // must re-arm the short fuse rather than letting a mid-flurry gap disengage early.
+            val job = launchReaction()
+
+            emit(primaryAddress, ConversationAwarenessEvent.START)
+            emit(primaryAddress, ConversationAwarenessEvent.HOLD)
+            advanceTimeBy(windDownTimeoutMs * 2 / 3)
+            runCurrent()
+            emit(primaryAddress, ConversationAwarenessEvent.HOLD) // re-arms
+            advanceTimeBy(windDownTimeoutMs * 2 / 3) // <1 fuse since the last HOLD
+            runCurrent()
+            verify(exactly = 0) { mediaControl.restoreMusicVolume(any()) }
+
+            advanceTimeBy(windDownTimeoutMs) // now >1 fuse since the last frame
+            runCurrent()
+            verify(exactly = 1) { mediaControl.restoreMusicVolume(10) }
+            job.cancel()
+        }
+
+    @Test
+    fun `fresh START during a wind-down cancels the short fuse`() = runTest(UnconfinedTestDispatcher()) {
+        // Wearer resumes speaking while the wind-down fuse is burning: the START must switch back
+        // to the long backstop — otherwise media resumes ~6s into the renewed conversation.
         val job = launchReaction()
 
         emit(primaryAddress, ConversationAwarenessEvent.START)
-        advanceTimeBy(staleTimeoutMs * 7 / 10)
+        emit(primaryAddress, ConversationAwarenessEvent.HOLD) // wind-down begins, short fuse armed
+        emit(primaryAddress, ConversationAwarenessEvent.START) // speaking again
+        advanceTimeBy(windDownTimeoutMs * 3)
         runCurrent()
-        emit(primaryAddress, ConversationAwarenessEvent.HOLD) // resets the backstop
-        advanceTimeBy(staleTimeoutMs * 7 / 10) // <1 backstop since the HOLD — still engaged
-        runCurrent()
-        verify(exactly = 0) { mediaControl.restoreMusicVolume(any()) }
 
-        advanceTimeBy(staleTimeoutMs / 2) // now >1 backstop since the last frame
+        verify(exactly = 0) { mediaControl.restoreMusicVolume(any()) }
+        job.cancel()
+    }
+
+    @Test
+    fun `PAUSE with dropped terminal frame resumes via the wind-down fuse`() = runTest(UnconfinedTestDispatcher()) {
+        // Same #608 single-pod scenario but with the PAUSE action: the fuse-driven disengage goes
+        // through the resume guards (age window, worn, already-playing) — at ~6s the pause is well
+        // inside PAUSE_RESUME_WINDOW, so media must resume rather than stay paused indefinitely.
+        devicesFlow.value = listOf(mockPodDevice(primaryAddress, ConversationAction.PAUSE))
+        val job = launchReaction()
+
+        emit(primaryAddress, ConversationAwarenessEvent.START)
+        coVerify(exactly = 1) { mediaControl.sendPause(false) }
+        emit(primaryAddress, ConversationAwarenessEvent.HOLD) // wind-down begins, terminal dropped
+        coVerify(exactly = 0) { mediaControl.sendPlay() }
+
+        advanceTimeBy(windDownTimeoutMs + 500)
         runCurrent()
-        verify(exactly = 1) { mediaControl.restoreMusicVolume(10) }
+        coVerify(exactly = 1) { mediaControl.sendPlay() }
         job.cancel()
     }