feat/refactor-ab-connector

androidApp/src/main/kotlin/pl/llp/aircasting/service/AirBeamForegroundService.kt

@@ -8,9 +8,9 @@ import android.app.Service
 import android.content.Context
 import android.content.Intent
 import android.content.pm.ServiceInfo
-import android.os.Build
 import android.os.IBinder
 import androidx.core.app.NotificationCompat
+import androidx.core.app.ServiceCompat
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
@@ -20,12 +20,16 @@ import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
 import org.koin.core.component.KoinComponent
 import org.koin.core.component.inject
-import pl.llp.aircasting.bluetooth.AirBeamSessionController
-import pl.llp.aircasting.bluetooth.SessionState
+import pl.llp.aircasting.bluetooth.AirBeamLink
+import pl.llp.aircasting.bluetooth.LinkState
 
+/**
+ * Keeps the process alive while the AirBeam link is up. It holds no connection itself — the
+ * shared [AirBeamLink] does; this only mirrors its state and stops when the link ends.
+ */
 class AirBeamForegroundService : Service(), KoinComponent {
 
-  private val sessionController: AirBeamSessionController by inject()
+  private val link: AirBeamLink by inject()
   private val serviceScope = CoroutineScope(Dispatchers.Main + SupervisorJob())
 
   companion object {
@@ -34,12 +38,7 @@ class AirBeamForegroundService : Service(), KoinComponent {
     private const val NOTIFICATION_ID = 1001
 
     fun start(context: Context) {
-      val intent = Intent(context, AirBeamForegroundService::class.java)
-      if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.O) {
-        context.startForegroundService(intent)
-      } else {
-        context.startService(intent)
-      }
+      context.startForegroundService(Intent(context, AirBeamForegroundService::class.java))
     }
 
     fun stop(context: Context) {
@@ -51,13 +50,17 @@ class AirBeamForegroundService : Service(), KoinComponent {
   override fun onCreate() {
     super.onCreate()
     createNotificationChannel()
-    observeSessionState()
+    observeLink()
   }
 
   override fun onStartCommand(intent: Intent?, flags: Int, startId: Int): Int {
     val initialNotification = buildNotification("AirBeam service active")
-    startForegroundWithCompat(initialNotification)
-    return START_STICKY
+    // ServiceCompat drops the service type below API 29, where the overload doesn't exist.
+    ServiceCompat.startForeground(
+      this, NOTIFICATION_ID, initialNotification, ServiceInfo.FOREGROUND_SERVICE_TYPE_CONNECTED_DEVICE,
+    )
+    // A restarted process has an idle link and nothing to restore it from — don't come back.
+    return START_NOT_STICKY
   }
 
   override fun onBind(intent: Intent?): IBinder? = null
@@ -67,35 +70,27 @@ class AirBeamForegroundService : Service(), KoinComponent {
     super.onDestroy()
   }
 
-  private fun observeSessionState() {
-    sessionController.state
-      .dropWhile { it is SessionState.Idle }
+  private fun observeLink() {
+    link.state
+      .dropWhile { it is LinkState.Idle }
       .onEach { state ->
         when (state) {
-          SessionState.Idle, is SessionState.Failed -> {
+          LinkState.Idle, is LinkState.Failed -> {
             stopForeground(STOP_FOREGROUND_REMOVE)
             stopSelf()
           }
-          is SessionState.Connecting -> updateNotification("Connecting to ${state.target.name}...")
-          is SessionState.Connected -> updateNotification("Connected to AirBeam")
-          is SessionState.Configuring -> updateNotification("Configuring session...")
-          is SessionState.Recording -> updateNotification("Recording active session")
-          is SessionState.Reconnecting -> updateNotification("Reconnecting to ${state.target.name} (Attempt ${state.attempt})...")
+          is LinkState.Connecting -> updateNotification("Connecting to ${state.target.name}...")
+          is LinkState.Connected -> updateNotification("Connected to ${state.target.name}")
+          is LinkState.Reconnecting -> updateNotification("Reconnecting to ${state.target.name} (Attempt ${state.attempt})...")
         }
       }
       .launchIn(serviceScope)
   }
 
   private fun createNotificationChannel() {
-    if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.O) {
-      val channel = NotificationChannel(
-        CHANNEL_ID,
-        CHANNEL_NAME,
-        NotificationManager.IMPORTANCE_LOW
-      )
-      val manager = getSystemService(NOTIFICATION_SERVICE) as NotificationManager
-      manager.createNotificationChannel(channel)
-    }
+    val channel = NotificationChannel(CHANNEL_ID, CHANNEL_NAME, NotificationManager.IMPORTANCE_LOW)
+    val manager = getSystemService(NOTIFICATION_SERVICE) as NotificationManager
+    manager.createNotificationChannel(channel)
   }
 
   private fun buildNotification(contentText: String): Notification {
@@ -114,16 +109,4 @@ class AirBeamForegroundService : Service(), KoinComponent {
     val manager = getSystemService(NOTIFICATION_SERVICE) as NotificationManager
     manager.notify(NOTIFICATION_ID, notification)
   }
-
-  private fun startForegroundWithCompat(notification: Notification) {
-    if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.Q) {
-      startForeground(
-        NOTIFICATION_ID,
-        notification,
-        ServiceInfo.FOREGROUND_SERVICE_TYPE_CONNECTED_DEVICE
-      )
-    } else {
-      startForeground(NOTIFICATION_ID, notification)
-    }
-  }
 }

docs/DOMAIN_KNOWLEDGE.md

@@ -118,6 +118,17 @@ Enums:
    `DISCONNECTED`; a 2s-interval reconnect loop starts (no hard cap).
 6. **Resume** — reconnect success → `RECORDING`. User can also disconnect deliberately
    (standalone mode) keeping the session.
+   **Reconnect rules, exactly** (`AirBeamService.onMessageEvent(SensorDisconnectedUnexpectedlyEvent)`,
+   `AirBeamReconnector`): only for a **mobile** session (fixed sessions record on the device);
+   not in standalone mode; one loop at a time; every **2 s**, **no cap** — it ends on
+   success, user stop/disconnect, or sync start. On success the device is **resumed, not
+   reconfigured** (`reconnectMobileSession()`): AB3/Mini V1 get the mobile-mode config again;
+   AB2 gets its mobile-mode message; **Mini V2 keeps running its session through the drop** —
+   it gets a time sync, then `ContinueSession` if it reports `HasSavedSession`, nothing if
+   `Running`; legacy logs and carries on for any other state. Re-sending a V2 session config
+   would be wrong (it would be rejected or start a second session).
+   In the rewrite these live in `AirBeamLink` (connection) — whether a drop means
+   `DISCONNECTED` or a finished session is the recording side's call.
 7. **Finish** — `StopRecordingEvent` → final averaging *(dropped in the rewrite — §7)*, delete active-session rows, status
    `FINISHED`, backend sync. All services/connectors self-tear-down.
 
@@ -178,6 +189,31 @@ Path A (live V1) is the older variant: it stamps `LocationHelper.lastLocation()`
 
 Fixed-session measurements carry no per-measurement location (it lives on the session).
 
+### Live readings off the device — **KEEP** (as implemented in the rewrite, 2026-10-20)
+
+What `AirBeamConnection.readings` emits (`SensorReading`: sensor name, unit symbol, value,
+device time or none). Sensor name + unit are the stream identity the v3 create call sends.
+
+| Device | Source | Format | Time |
+|--------|--------|--------|------|
+| AB3 | notify `ffe1` temp F, `ffe3` humidity, `ffe4` PM1, `ffe5` PM2.5, `ffe6` PM10 (`AirBeam3Configurator`) | ASCII line, below | none — stamp on receipt |
+| Mini V1 | notify `ffe4`, `ffe5` (`AirBeamMiniConfigurator`; its `ffe7` sends only battery `"85%"`) | ASCII line | none |
+| AB2 | RFCOMM socket, one line per reading (`AirBeam2Reader`) | ASCII line | none |
+| Mini V2 | indicate `…0005` live, `…0006` sync backlog | binary LE (`ble_mobile_app_guide.md` §6) | device epoch seconds |
+
+- **ASCII line** (`ResponseParser`): `value;package;sensor name;type;short type;unit name;
+  unit symbol;T1;T2;T3;T4;T5` — 12 fields, the five thresholds integers. Anything else
+  (battery text, torn line) is dropped.
+- **Mini V2 names** carry no metadata on the wire; legacy uploads them as
+  `AirBeamMini-PM1` / `AirBeamMini-PM2.5`, unit `µg/m³`
+  (`AirBeamMiniV2Configurator.postLiveMeasurementEvents`). Keep these exact names so the
+  backend streams line up with legacy-recorded sessions.
+- **MTU** (Android only; iOS negotiates itself): V1 BLE devices request 517
+  (`SyncableAirBeamConfigurator.MAX_MTU`) — a line doesn't fit the default 20-byte payload;
+  Mini V2 requests 247 (fits a 244-byte sync chunk). A refused request is logged, not fatal.
+- The V2 sync backlog flows through the same `readings` as live data, so the recording side
+  has one save path (dedup on `(stream, time)` makes replays harmless).
+
 UI consumes the DB reactively: Room `LiveData` → repository projections → `SessionsObserver`
 (diff engine) → dashboard cards. DB is the intended single source of truth.
 

docs/airbeam/02-legacy-connection-internals.md

@@ -243,6 +243,12 @@ only** → `tryToReconnectPeriodically` (`AirBeamService.kt:123-140`). Retries e
 
 - **KEEP (behavior):** connection outlives the UI; mobile auto-reconnect; distinguish
   user-disconnect from unexpected drop (your `ConnectionStatus.DisconnectedUnexpectedly`).
+- **Rewrite status (2026-10-20):** `AirBeamLink` (`bluetooth/AirBeamLink.kt`) is the owner —
+  a Koin singleton running in a process-lifetime scope, so the connection outlives screens.
+  It carries the reconnect rules above (`DOMAIN_KNOWLEDGE.md` §2, step 6) and is unit-tested
+  against fakes (`AirBeamLinkTest`). `AirBeamForegroundService` only mirrors `LinkState` to keep
+  the process alive; nothing starts it yet (the recording engine will, behind a
+  `BackgroundSessionHost`).
 - **KMP mapping:** ownership is the `01-connecting.md` §9 question. Android → foreground service
   (`expect`/`actual` + Android `FOREGROUND_SERVICE_CONNECTED_DEVICE`). iOS → CoreBluetooth
   background mode `bluetooth-central` + state restoration. Reconnect loop → a clean Flow + backoff

shared/src/androidMain/kotlin/pl/llp/aircasting/bluetooth/transport/ble/Mtu.android.kt

@@ -0,0 +1,16 @@
+package pl.llp.aircasting.bluetooth.transport.ble
+
+import co.touchlab.kermit.Logger
+import com.juul.kable.ServicesDiscoveredPeripheral
+import kotlinx.coroutines.CancellationException
+
+internal actual suspend fun ServicesDiscoveredPeripheral.requestMtuIfSupported(mtu: Int) {
+  try {
+    val negotiated = requestMtu(mtu)
+    Logger.withTag("BleMtu").d { "MTU negotiated: $negotiated" }
+  } catch (cancel: CancellationException) {
+    throw cancel
+  } catch (e: Exception) {
+    Logger.withTag("BleMtu").w(e) { "MTU request for $mtu refused" }
+  }
+}

shared/src/androidMain/kotlin/pl/llp/aircasting/bluetooth/transport/classic/ClassicAirBeamConnector.kt

@@ -15,29 +15,34 @@ import pl.llp.aircasting.bluetooth.AirBeamConnection
 import pl.llp.aircasting.bluetooth.AirBeamConnector
 import pl.llp.aircasting.bluetooth.AirBeamDevice
 import pl.llp.aircasting.bluetooth.ConfigResult
+import pl.llp.aircasting.bluetooth.ConnectResult
 import pl.llp.aircasting.bluetooth.ConnectionStatus
 import pl.llp.aircasting.bluetooth.DeviceId
 import pl.llp.aircasting.bluetooth.DiscoveredAirBeam
 import pl.llp.aircasting.bluetooth.FailureReason
+import pl.llp.aircasting.bluetooth.SensorReading
 import pl.llp.aircasting.bluetooth.SessionConfig
 import pl.llp.aircasting.bluetooth.Transport
 import pl.llp.aircasting.bluetooth.protocol.AirBeamProtocol
 import pl.llp.aircasting.bluetooth.transport.accumulateDistinct
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
-import kotlinx.coroutines.TimeoutCancellationException
 import kotlinx.coroutines.cancel
+import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.awaitClose
+import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.flow.callbackFlow
+import kotlinx.coroutines.flow.receiveAsFlow
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.runInterruptible
 import kotlinx.coroutines.withContext
-import kotlinx.coroutines.withTimeout
 import java.io.IOException
 import java.util.UUID
 import kotlin.let
@@ -91,63 +96,71 @@ class ClassicAirBeamConnector(
   }.accumulateDistinct()
 
   @SuppressLint("MissingPermission")
-  override suspend fun connect(target: DiscoveredAirBeam): AirBeamConnection {
-    val adapter = adapter ?: return failed(FailureReason.RadioOrPermissionMissing)
+  override suspend fun connect(target: DiscoveredAirBeam): ConnectResult {
+    val adapter = adapter ?: return ConnectResult.Failed(FailureReason.RadioOrPermissionMissing)
     adapter.cancelDiscovery() // inquiry saturates the radio — always stop before connecting
 
     val socket = adapter.getRemoteDevice(target.id.value)
       .createRfcommSocketToServiceRecord(SPP_UUID)
 
-    return withContext(Dispatchers.IO) {
+    return coroutineScope {
+      // socket.connect() blocks and ignores cancellation; closing the socket is what aborts it.
+      val watchdog = launch {
+        delay(CONNECT_TIMEOUT_MS.milliseconds)
+        socket.safeClose()
+      }
       try {
-        withTimeout(CONNECT_TIMEOUT_MS.milliseconds) {
-          socket.connect()      // blocking until linked or IOException
-        }
+        runInterruptible(Dispatchers.IO) { socket.connect() }
+        watchdog.cancel()
         delay(ESTIMATED_CONNECTING_TIME_MS.milliseconds)
-        ClassicConnection(socket, target.device)
-      } catch (e: TimeoutCancellationException) {
-        socket.safeClose()
-        failed(FailureReason.LinkTimeout)
+        ConnectResult.Connected(ClassicConnection(socket, target.device))
       } catch (_: IOException) {
+        val timedOut = watchdog.isCompleted // ran to the end: it closed the socket on us
+        watchdog.cancel()
         socket.safeClose()
-        failed(FailureReason.HandshakeFailed)
+        ConnectResult.Failed(if (timedOut) FailureReason.LinkTimeout else FailureReason.HandshakeFailed)
+      } catch (cancel: CancellationException) {
+        socket.safeClose()
+        throw cancel
       }
     }
   }
 }
 
 private class ClassicConnection(
   private val socket: BluetoothSocket,
-  val device: AirBeamDevice,
+  override val device: AirBeamDevice,
 ) : AirBeamConnection {
   private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())
-  private val _status = MutableStateFlow<ConnectionStatus>(ConnectionStatus.Ready(device))
+  private val _status = MutableStateFlow(ConnectionStatus.Ready)
   override val status: StateFlow<ConnectionStatus> = _status.asStateFlow()
   override val deviceState = null // AB2 does not report its own state
   private val log = Logger.withTag("ClassicConnection")
-  private var isDisconnectingExplicitly = false
+  private val received = Channel<SensorReading>(Channel.UNLIMITED)
+  override val readings = received.receiveAsFlow()
 
   init {
+    // AB2 streams one ASCII measurement line at a time (legacy AirBeam2Reader).
     scope.launch {
-      val buffer = ByteArray(1024)
       try {
-        while (!isDisconnectingExplicitly && socket.isConnected) {
-          val bytesRead = socket.inputStream.read(buffer)
-          if (bytesRead == -1) break
+        val lines = socket.inputStream.bufferedReader()
+        while (true) {
+          val line = runInterruptible { lines.readLine() } ?: break
+          AirBeamProtocol.Standard.parseReading(line)?.let { received.trySend(it) }
         }
       } catch (e: IOException) {
         log.d { "Classic socket read loop ended: ${e.message}" }
       } finally {
-        if (!isDisconnectingExplicitly && _status.value !is ConnectionStatus.Disconnected) {
+        received.close()
+        if (_status.compareAndSet(ConnectionStatus.Ready, ConnectionStatus.DisconnectedUnexpectedly)) {
           log.w { "Classic connection dropped unexpectedly" }
-          _status.value = ConnectionStatus.DisconnectedUnexpectedly
         }
       }
     }
   }
 
   override suspend fun configure(config: SessionConfig): ConfigResult {
-    log.i { "Configuring AirBeam2 ($device) with config: $config" }
+    log.i { "Configuring AirBeam2 ($device) with $config" }
     return try {
       when (config) {
         is SessionConfig.Mobile -> configureMobile()
@@ -160,6 +173,9 @@ private class ClassicConnection(
     }
   }
 
+  /** Legacy `AirBeam2Configurator.reconnectMobileSession`: mobile mode again. */
+  override suspend fun resumeMobileSession(config: SessionConfig.Mobile): ConfigResult = configure(config)
+
   private suspend fun configureMobile(): ConfigResult {
     // Mobile Session (Bluetooth): Send [0x01, 0xFF]
     sendCommand(AirBeamProtocol.AirBeam2.startMobileCommand())
@@ -210,20 +226,12 @@ private class ClassicConnection(
   }
 
   override suspend fun disconnect() {
-    isDisconnectingExplicitly = true
+    _status.compareAndSet(ConnectionStatus.Ready, ConnectionStatus.Disconnected)
     scope.cancel()
     withContext(Dispatchers.IO) { socket.safeClose() }
-    _status.value = ConnectionStatus.Disconnected
   }
 }
 
-private fun failed(reason: FailureReason): AirBeamConnection = object : AirBeamConnection {
-  override val status = MutableStateFlow<ConnectionStatus>(ConnectionStatus.Failed(reason))
-  override val deviceState = null
-  override suspend fun configure(config: SessionConfig) = ConfigResult.UnknownFailure
-  override suspend fun disconnect() {}
-}
-
 private fun BluetoothSocket.safeClose() {
   try {
     close()