fix(service): stop a stale saved address overwriting a newer selection (#7329)

core/service/src/commonMain/kotlin/org/meshtastic/core/service/SharedRadioInterfaceService.kt

@@ -198,6 +198,11 @@ class SharedRadioInterfaceService(
     private val _currentDeviceAddressFlow = MutableStateFlow<String?>(radioPrefs.devAddr.value)
     override val currentDeviceAddressFlow: StateFlow<String?> = _currentDeviceAddressFlow.asStateFlow()
 
+    private val selectionLock = SynchronizedObject()
+
+    /** Set by the first [setDeviceAddress]; from then on this process owns the selection, not the prefs mirror. */
+    private var selectionPublished = false
+
     // Monotonically increasing generation bumped on every transport start (including same-address reconnect). Exposed
     // through [sessionGeneration] so the controller layer can clear connection-session identity at each session
     // boundary instead of only when the selected address changes.
@@ -571,11 +576,12 @@ class SharedRadioInterfaceService(
         // starts a transport. Transport start remains driven exclusively by connect() (initial),
         // setDeviceAddress() (explicit user switch), BLE/network state changes (environment
         // recovery), and liveness restarts (zombie recovery) — see startTransportLocked() callers.
-        // _currentDeviceAddressFlow is a MutableStateFlow (atomic .value), so the unconditional
-        // assignment here is race-free without holding transportMutex; same-address writes are
-        // idempotent no-ops.
+        // It stops at the first setDeviceAddress(): prefs commit asynchronously, so a later emission can carry an
+        // older selection than the one already published.
         radioPrefs.devAddr
-            .onEach { addr -> _currentDeviceAddressFlow.value = addr }
+            .onEach { addr ->
+                synchronized(selectionLock) { if (!selectionPublished) _currentDeviceAddressFlow.value = addr }
+            }
             .catch { Logger.e(it) { "radioPrefs.devAddr address-sync flow crashed" } }
             .launchIn(processLifecycle.coroutineScope)
     }
@@ -834,7 +840,10 @@ class SharedRadioInterfaceService(
 
         Logger.d { "Setting bonded device to ${sanitized?.anonymize}" }
         radioPrefs.setDevAddr(sanitized)
-        _currentDeviceAddressFlow.value = sanitized
+        synchronized(selectionLock) {
+            selectionPublished = true
+            _currentDeviceAddressFlow.value = sanitized
+        }
 
         processLifecycle.coroutineScope.launch {
             transportMutex.withLock {

core/service/src/commonTest/kotlin/org/meshtastic/core/service/SharedRadioInterfaceServiceLivenessTest.kt

@@ -53,6 +53,7 @@ import org.meshtastic.core.network.repository.NetworkRepository
 import org.meshtastic.core.network.repository.SerialDevicePresence
 import org.meshtastic.core.repository.PlatformAnalytics
 import org.meshtastic.core.repository.RadioInterfaceService
+import org.meshtastic.core.repository.RadioPrefs
 import org.meshtastic.core.repository.RadioTransport
 import org.meshtastic.core.repository.RadioTransportFactory
 import org.meshtastic.core.repository.TransportDisconnectReason
@@ -287,6 +288,7 @@ class SharedRadioInterfaceServiceLivenessTest {
         transportProvider: () -> RadioTransport = { FakeRadioTransport().also { createdTransports.add(it) } },
         networkAvailability: MutableStateFlow<Boolean> = MutableStateFlow(true),
         startConnected: Boolean = true,
+        radioPrefs: RadioPrefs = this.radioPrefs,
     ): SharedRadioInterfaceService {
         every { networkRepository.networkAvailable } returns networkAvailability
         every { networkRepository.resolvedList } returns MutableSharedFlow()
@@ -452,6 +454,59 @@ class SharedRadioInterfaceServiceLivenessTest {
             }
         }
 
+    /** Persists like DataStore: a write lands only when the test commits it, in order. */
+    private class DeferredRadioPrefs(saved: String?) : RadioPrefs {
+        override val devAddr = MutableStateFlow(saved)
+        override val devName = MutableStateFlow<String?>(null)
+        private val pending = ArrayDeque<String?>()
+
+        override fun setDevAddr(address: String?) {
+            pending.addLast(address)
+        }
+
+        override fun setDevName(name: String?) {
+            devName.value = name
+        }
+
+        fun commitThrough(address: String) {
+            do {
+                val next = pending.removeFirst()
+                devAddr.value = next
+            } while (next != address)
+        }
+    }
+
+    @Test
+    fun `a late commit of an older selection does not rewind the selected address`() = runTest(testDispatcher) {
+        val prefs = DeferredRadioPrefs(saved = "xAA:AA:AA:AA:AA:AA")
+        val service = createConnectedService("xAA:AA:AA:AA:AA:AA", startConnected = false, radioPrefs = prefs)
+        try {
+            service.setDeviceAddress("xBB:BB:BB:BB:BB:BB")
+            service.setDeviceAddress("xCC:CC:CC:CC:CC:CC")
+
+            prefs.commitThrough("xBB:BB:BB:BB:BB:BB")
+            testDispatcher.scheduler.runCurrent()
+
+            assertEquals("xCC:CC:CC:CC:CC:CC", service.currentDeviceAddressFlow.value)
+        } finally {
+            service.disconnect()
+        }
+    }
+
+    @Test
+    fun `the saved address still reaches the flow when it loads after construction`() = runTest(testDispatcher) {
+        val prefs = DeferredRadioPrefs(saved = null)
+        val service = createConnectedService("xAA:AA:AA:AA:AA:AA", startConnected = false, radioPrefs = prefs)
+        try {
+            prefs.devAddr.value = "xAA:AA:AA:AA:AA:AA"
+            testDispatcher.scheduler.runCurrent()
+
+            assertEquals("xAA:AA:AA:AA:AA:AA", service.currentDeviceAddressFlow.value)
+        } finally {
+            service.disconnect()
+        }
+    }
+
     @Test
     fun `setDeviceAddress contains factory failure and same-address repair can retry`() = runTest(testDispatcher) {
         bluetoothRepository.setBluetoothEnabled(false)