Harden idleness tracking and cooldown against races

Three fixes from code review:

- EventDispatcher: serialize trackedJobs add/remove with _isIdle writes via a synchronized block. Without this, a stale completion callback could overwrite the _isIdle=false set by a concurrent trackJob, leaving the orchestrator thinking the dispatcher was idle while a job was still tracked.

- MonitorOrchestrator: replace the StateFlow-conflation-vulnerable withTimeoutOrNull/filter pattern with a deterministic workGeneration counter snapshot before delay(15s) compared after. A fast dispatch finishing inside the grace window now reliably restarts the cooldown.

- VolumeTool: defensive guard for degenerate stream bounds (min > max) before coerceIn, which would throw IllegalArgumentException on an empty range.

Author: d4rken

app/src/main/java/eu/darken/bluemusic/monitor/core/audio/VolumeTool.kt

@@ -152,6 +152,10 @@ class VolumeTool @Inject constructor(
 
         val max = getMaxVolume(streamId)
         val min = getMinVolume(streamId)
+        if (min > max) {
+            log(TAG, WARN) { "Invalid stream bounds: min=$min > max=$max for $streamId; aborting changeVolume." }
+            return false
+        }
         val clampedTarget = targetLevel.coerceIn(min, max)
         if (clampedTarget != targetLevel) {
             log(TAG, WARN) { "Target level $targetLevel clamped to $clampedTarget (min=$min, max=$max)." }

app/src/main/java/eu/darken/bluemusic/monitor/core/service/EventDispatcher.kt

@@ -34,6 +34,7 @@ import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.atomic.AtomicLong
 import javax.inject.Inject
 import javax.inject.Singleton
 
@@ -68,16 +69,32 @@ class EventDispatcher @Inject constructor(
     private val _isIdle = MutableStateFlow(true)
     val isIdle: StateFlow<Boolean> = _isIdle.asStateFlow()
 
+    // Monotonic counter incremented on every job tracked. Allows the orchestrator's
+    // cooldown to detect "work happened during grace" deterministically, without
+    // depending on StateFlow conflation behaviour for short-lived work.
+    private val workGeneration = AtomicLong(0)
+    fun currentWorkGeneration(): Long = workGeneration.get()
+
+    // Lock that serializes the (set mutation + _isIdle write) pair so completion
+    // callbacks for old jobs can't overwrite an `_isIdle = false` set by a fresh
+    // trackJob — see EventDispatcherTest.`isIdle stays consistent under concurrent track and complete`.
+    private val trackingLock = Any()
+
     suspend fun awaitIdle() {
         isIdle.filter { it }.first()
     }
 
     private fun trackJob(job: Job) {
-        trackedJobs.add(job)
-        _isIdle.value = false
+        workGeneration.incrementAndGet()
+        synchronized(trackingLock) {
+            trackedJobs.add(job)
+            _isIdle.value = false
+        }
         job.invokeOnCompletion {
-            trackedJobs.remove(job)
-            if (trackedJobs.isEmpty()) _isIdle.value = true
+            synchronized(trackingLock) {
+                trackedJobs.remove(job)
+                if (trackedJobs.isEmpty()) _isIdle.value = true
+            }
         }
     }
 

app/src/main/java/eu/darken/bluemusic/monitor/core/service/MonitorOrchestrator.kt

@@ -22,13 +22,10 @@ import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.catch
 import kotlinx.coroutines.flow.distinctUntilChanged
 import kotlinx.coroutines.flow.emptyFlow
-import kotlinx.coroutines.flow.filter
-import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.flatMapLatest
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
-import kotlinx.coroutines.withTimeoutOrNull
 import java.time.Duration
 import javax.inject.Inject
 import javax.inject.Singleton
@@ -121,16 +118,15 @@ class MonitorOrchestrator @Inject constructor(
                         log(TAG) { "Active devices present; waiting for dispatcher idle then 15s grace before stopping." }
                         while (true) {
                             eventDispatcher.awaitIdle()
-                            val wentBusy = withTimeoutOrNull(Duration.ofSeconds(15).toMillis()) {
-                                eventDispatcher.isIdle.filter { !it }.first()
-                                true
+                            val genAtStart = eventDispatcher.currentWorkGeneration()
+                            delay(Duration.ofSeconds(15).toMillis())
+                            if (eventDispatcher.currentWorkGeneration() != genAtStart) {
+                                log(TAG) { "Dispatcher had new work during grace; restarting cooldown." }
+                                continue
                             }
-                            if (wentBusy == null) {
-                                log(TAG) { "Dispatcher idle for 15s; stopping." }
-                                monitorJob.cancel()
-                                return@flow
-                            }
-                            log(TAG) { "Dispatcher went busy during grace; restarting cooldown." }
+                            log(TAG) { "Dispatcher idle for 15s; stopping." }
+                            monitorJob.cancel()
+                            return@flow
                         }
                     }
 

app/src/test/java/eu/darken/bluemusic/monitor/core/audio/VolumeToolTest.kt

@@ -142,6 +142,31 @@ class VolumeToolTest : BaseTest() {
         volumeTool.hasRecentTarget(AudioStream.Id.STREAM_MUSIC, 7) shouldBe true
     }
 
+    @Test
+    fun `min greater than max returns false without writing or recording target`() = runTest {
+        // Defensive guard: if the platform reports degenerate stream bounds (min > max),
+        // VolumeTool aborts before calling coerceIn (which would throw IllegalArgumentException).
+        every { audioManager.getStreamMaxVolume(any()) } returns -1
+        // getMinVolume returns 0 in unit tests (Build.VERSION.SDK_INT==0 path), so 0 > -1.
+        audioLevels[AudioStream.Id.STREAM_MUSIC] = 0
+        val writes = mutableListOf<Int>()
+        every { audioManager.setStreamVolume(AudioStream.Id.STREAM_MUSIC.id, any(), any()) } answers {
+            val level = secondArg<Int>()
+            audioLevels[AudioStream.Id.STREAM_MUSIC] = level
+            writes += level
+        }
+
+        val result = volumeTool.changeVolume(
+            streamId = AudioStream.Id.STREAM_MUSIC,
+            targetLevel = 5,
+            delay = Duration.ofMillis(1),
+        )
+
+        result shouldBe false
+        writes shouldBe emptyList()
+        volumeTool.hasRecentTarget(AudioStream.Id.STREAM_MUSIC, 5) shouldBe false
+    }
+
     @Test
     fun `target level below min is clamped to min`() = runTest {
         // In unit tests Build.VERSION.SDK_INT is 0 (no Robolectric), so getMinVolume returns 0.

app/src/test/java/eu/darken/bluemusic/monitor/core/service/EventDispatcherTest.kt

@@ -969,4 +969,74 @@ class EventDispatcherTest : BaseTest() {
         // Release the deferred so the cancelled coroutine can finish unwinding cleanly.
         release.complete(Unit)
     }
+
+    @Test
+    fun `currentWorkGeneration starts at 0 and increments on dispatch`() = runTest {
+        val buds = managedDevice(budsAddress, connected = true)
+        devicesFlow.value = listOf(buds)
+
+        val release = CompletableDeferred<Unit>()
+        coEvery { module1.handle(any()) } coAnswers { release.await() }
+
+        val dispatcher = createDispatcher()
+        dispatcher.currentWorkGeneration() shouldBe 0L
+
+        dispatcher.dispatch(event(budsAddress, CONNECTED))
+        advanceUntilIdle()
+
+        // At least one trackJob ran (the cancellable launch).
+        (dispatcher.currentWorkGeneration() > 0L) shouldBe true
+
+        release.complete(Unit)
+    }
+
+    @Test
+    fun `currentWorkGeneration keeps increasing across multiple dispatches`() = runTest {
+        val buds = managedDevice(budsAddress, connected = true)
+        devicesFlow.value = listOf(buds)
+
+        val firstRelease = CompletableDeferred<Unit>()
+        val secondRelease = CompletableDeferred<Unit>()
+        var callCount = 0
+        coEvery { module1.handle(any()) } coAnswers {
+            val n = ++callCount
+            if (n == 1) firstRelease.await() else secondRelease.await()
+        }
+
+        val dispatcher = createDispatcher()
+
+        dispatcher.dispatch(event(budsAddress, CONNECTED))
+        advanceUntilIdle()
+        val genAfterFirst = dispatcher.currentWorkGeneration()
+        (genAfterFirst > 0L) shouldBe true
+
+        dispatcher.dispatch(event(budsAddress, DISCONNECTED))
+        advanceUntilIdle()
+        val genAfterSecond = dispatcher.currentWorkGeneration()
+        (genAfterSecond > genAfterFirst) shouldBe true
+
+        firstRelease.complete(Unit)
+        secondRelease.complete(Unit)
+    }
+
+    @Test
+    fun `currentWorkGeneration increments even when a fast dispatch finishes before isIdle is observed`() = runTest {
+        // Verifies that the orchestrator's "did anything happen during grace?" check
+        // remains accurate even for blink-and-you-miss-it dispatches whose isIdle=false
+        // gets conflated by StateFlow before any collector observes it.
+        val buds = managedDevice(budsAddress, connected = true)
+        devicesFlow.value = listOf(buds)
+
+        // Module returns immediately — total in-flight time is microseconds.
+        coEvery { module1.handle(any()) } returns Unit
+
+        val dispatcher = createDispatcher()
+        val genBefore = dispatcher.currentWorkGeneration()
+
+        dispatcher.dispatch(event(budsAddress, CONNECTED))
+        advanceUntilIdle()
+
+        dispatcher.isIdle.value shouldBe true
+        (dispatcher.currentWorkGeneration() > genBefore) shouldBe true
+    }
 }

app/src/test/java/eu/darken/bluemusic/monitor/core/service/MonitorOrchestratorTest.kt

@@ -25,6 +25,7 @@ import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
 import testhelpers.BaseTest
 import java.time.Duration
+import java.util.concurrent.atomic.AtomicLong
 
 class MonitorOrchestratorTest : BaseTest() {
 
@@ -41,6 +42,7 @@ class MonitorOrchestratorTest : BaseTest() {
     private lateinit var devicesFlow: MutableStateFlow<List<ManagedDevice>>
     private lateinit var stateFlow: MutableStateFlow<BluetoothRepo.State>
     private lateinit var idleFlow: MutableStateFlow<Boolean>
+    private lateinit var workGen: AtomicLong
 
     @BeforeEach
     fun setup() {
@@ -58,9 +60,11 @@ class MonitorOrchestratorTest : BaseTest() {
         ringerModeObserver = mockk { every { ringerMode } returns MutableSharedFlow() }
         bluetoothEventQueue = mockk { every { events } returns MutableSharedFlow() }
         idleFlow = MutableStateFlow(true)
+        workGen = AtomicLong(0)
         eventDispatcher = mockk(relaxed = true) {
             every { isIdle } returns idleFlow
             coEvery { awaitIdle() } coAnswers { idleFlow.filter { it }.first() }
+            every { currentWorkGeneration() } answers { workGen.get() }
         }
         ringerModeTransitionHandler = mockk(relaxed = true)
         ownerRegistry = mockk(relaxed = true)
@@ -224,16 +228,25 @@ class MonitorOrchestratorTest : BaseTest() {
             monitorReturned = true
         }
 
-        // Idle from start; advance 10s into grace window
+        // Idle from start; advance 10s into the first grace window.
         advanceTimeBy(10_000)
         monitorReturned shouldBe false
 
-        // Mid-grace, dispatcher goes busy → resets cooldown
+        // Mid-grace: dispatcher gets new work — bump generation AND flip idle to false
+        // to simulate what trackJob() does in production. The cooldown checks generation
+        // (not isIdle) at the end of the 15s delay, so this *will* be detected.
+        workGen.incrementAndGet()
         idleFlow.value = false
+
+        // Finish out the original 15s delay — orchestrator sees generation changed, restarts.
+        advanceTimeBy(5_001)
+        monitorReturned shouldBe false
+
+        // Restart loop calls awaitIdle() which suspends because dispatcher is busy.
         advanceTimeBy(20_000)
         monitorReturned shouldBe false
 
-        // Dispatcher becomes idle again
+        // Dispatcher becomes idle again — awaitIdle() returns, new 15s grace begins.
         idleFlow.value = true
         advanceTimeBy(14_000)
         monitorReturned shouldBe false
@@ -245,6 +258,45 @@ class MonitorOrchestratorTest : BaseTest() {
         job.cancel()
     }
 
+    @Test
+    fun `work bump alone during grace restarts cooldown even without idle flip`() = runTest {
+        // Edge case: a fast dispatch that completes within the 15s grace window —
+        // its trackJob bumps the generation but isIdle may already be true again
+        // by the time the cooldown's delay completes. Generation-based check still detects it.
+        val device = managedDevice(
+            "AA:BB:CC:DD:EE:FF",
+            active = true,
+            requiresMonitor = false,
+        )
+        devicesFlow.value = listOf(device)
+
+        val orchestrator = createOrchestrator()
+        var monitorReturned = false
+
+        val job = launch {
+            orchestrator.monitor(this@runTest) {}
+            monitorReturned = true
+        }
+
+        // Mid-grace, simulate a brief dispatch that bumps generation but leaves idle=true.
+        advanceTimeBy(10_000)
+        workGen.incrementAndGet()
+
+        // Finish out the original 15s delay — generation differs → restart.
+        advanceTimeBy(5_001)
+        monitorReturned shouldBe false
+
+        // Second grace window with no further work.
+        advanceTimeBy(14_000)
+        monitorReturned shouldBe false
+
+        advanceTimeBy(2_000)
+        advanceUntilIdle()
+        monitorReturned shouldBe true
+
+        job.cancel()
+    }
+
     @Test
     fun `no devices connected - stops after 15s`() = runTest {
         devicesFlow.value = emptyList()