push feature stable

packages/flutter_mozilla_components/android/build.gradle

@@ -152,6 +152,7 @@ dependencies {
     //https://stackoverflow.com/questions/73782320/onbackinvokedcallback-is-not-enabled-for-the-application-in-set-androidenableo
     implementation 'androidx.activity:activity-ktx:1.13.0'
     implementation 'androidx.paging:paging-runtime-ktx:3.5.0'
+    implementation 'androidx.work:work-runtime:2.11.2'
 
     testImplementation("org.jetbrains.kotlin:kotlin-test")
     testImplementation("org.mockito:mockito-core:5.23.0")

packages/flutter_mozilla_components/android/src/main/kotlin/eu/weblibre/flutter_mozilla_components/ActiveProfile.kt

@@ -20,7 +20,12 @@
 package eu.weblibre.flutter_mozilla_components
 
 import android.content.Context
+import android.util.AtomicFile
 import java.io.File
+import java.io.FileNotFoundException
+import java.util.UUID
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
 
 object ActiveProfile {
     @Volatile
@@ -44,11 +49,50 @@ object ActiveProfile {
      * Resolve the active profile prefix from disk.
      * Called in Application.onCreate() to handle cold-start WorkManager scenarios.
      */
-    fun resolveFromDisk(context: Context) {
+    fun resolveFromDisk(context: Context): ProfileContext? = resolveContext(context)
+
+    /** Resolve the active profile without constructing browser components. */
+    @Synchronized
+    fun resolveContext(context: Context): ProfileContext? {
         val profileFile = File(context.filesDir, PwaConstants.CURRENT_PROFILE_FILE)
-        if (!profileFile.exists()) return
-        val uuid = profileFile.readText().trim().ifEmpty { return }
+        val uuid = try {
+            AtomicFile(profileFile).openRead().bufferedReader().use { it.readText() }.trim()
+        } catch (_: FileNotFoundException) {
+            return null
+        }.ifEmpty { return null }
         val relativePath = "${PwaConstants.PROFILES_DIR_NAME}/${PwaConstants.PROFILE_DIR_PREFIX}$uuid"
+        if (!File(context.filesDir, relativePath).isDirectory) return null
+        prefix = File(relativePath).name
+        return ProfileContext(context.applicationContext, relativePath)
+    }
+
+    /** Atomically select the profile used by the next browser process. */
+    @Synchronized
+    fun switchTo(context: Context, profileId: String) {
+        val normalizedId = UUID.fromString(profileId).toString()
+        require(normalizedId == profileId.lowercase()) { "Invalid profile id" }
+
+        val relativePath =
+            "${PwaConstants.PROFILES_DIR_NAME}/${PwaConstants.PROFILE_DIR_PREFIX}$normalizedId"
+        require(File(context.filesDir, relativePath).isDirectory) { "Profile does not exist" }
+
+        val profileFile = File(context.filesDir, PwaConstants.CURRENT_PROFILE_FILE)
+        profileFile.parentFile?.mkdirs()
+        val atomicFile = AtomicFile(profileFile)
+        val output = atomicFile.startWrite()
+        try {
+            output.write(normalizedId.toByteArray(Charsets.UTF_8))
+            atomicFile.finishWrite(output)
+        } catch (error: Throwable) {
+            atomicFile.failWrite(output)
+            throw error
+        }
         prefix = File(relativePath).name
     }
+
+    /** Prevent profile switches from crossing active-profile background work. */
+    internal suspend fun <T> withProfileLock(block: suspend () -> T): T =
+        profileMutex.withLock { block() }
+
+    private val profileMutex = Mutex()
 }

packages/flutter_mozilla_components/android/src/main/kotlin/eu/weblibre/flutter_mozilla_components/Components.kt

@@ -12,7 +12,7 @@ import eu.weblibre.flutter_mozilla_components.components.Core
 import eu.weblibre.flutter_mozilla_components.components.BackgroundServices
 import eu.weblibre.flutter_mozilla_components.components.Events
 import eu.weblibre.flutter_mozilla_components.components.Features
-import eu.weblibre.flutter_mozilla_components.components.Push
+import eu.weblibre.flutter_mozilla_components.push.Push
 import eu.weblibre.flutter_mozilla_components.components.Search
 import eu.weblibre.flutter_mozilla_components.components.Services
 import eu.weblibre.flutter_mozilla_components.components.UseCases
@@ -75,7 +75,11 @@ class Components(val profileApplicationContext: ProfileContext,
     }
     val features by lazy { Features(core.engine, core.store, addonEvents, tabContentEvents) }
     val search by lazy { Search(profileApplicationContext, core, useCases) }
-    val push by lazy { Push(this) }
+    private val pushDelegate = lazy { Push(this) }
+    val push: Push
+        get() = pushDelegate.value
+    internal val existingPush: Push?
+        get() = pushDelegate.takeIf { it.isInitialized() }?.value
 
     var mainBrowserEngineView: EngineView? = null
     var externalAppEngineView: EngineView? = null

packages/flutter_mozilla_components/android/src/main/kotlin/eu/weblibre/flutter_mozilla_components/FlutterMozillaComponentsPlugin.kt

@@ -11,6 +11,7 @@ import eu.weblibre.flutter_mozilla_components.api.GeckoEngineSettingsApiImpl
 import eu.weblibre.flutter_mozilla_components.feature.SandboxCaptureFeature
 import eu.weblibre.flutter_mozilla_components.pigeons.GeckoBrowserApi
 import eu.weblibre.flutter_mozilla_components.pigeons.GeckoEngineSettingsApi
+import eu.weblibre.flutter_mozilla_components.pigeons.GeckoPushApi
 
 import io.flutter.embedding.engine.plugins.FlutterPlugin
 import io.flutter.embedding.engine.plugins.activity.ActivityAware
@@ -43,7 +44,12 @@ class FlutterMozillaComponentsPlugin: FlutterPlugin, ActivityAware {
 
   override fun onDetachedFromEngine(binding: FlutterPlugin.FlutterPluginBinding) {
     SandboxCaptureFeature.detachFlutterEvents(binding.binaryMessenger)
+    GeckoPushApi.setUp(binding.binaryMessenger, null)
+    browserApi.disposePushApi()
     GlobalComponents.historyEvents = null
+    // The UnifiedPush receiver outlives the Flutter engine; without this it would keep dispatching
+    // onto a dead messenger. Failures are still retained on Push.lastError.
+    GlobalComponents.pushEvents = null
   }
 
   override fun onAttachedToActivity(binding: ActivityPluginBinding) {

packages/flutter_mozilla_components/android/src/main/kotlin/eu/weblibre/flutter_mozilla_components/GlobalComponents.kt

@@ -18,6 +18,7 @@ import eu.weblibre.flutter_mozilla_components.pigeons.GeckoAddonEvents
 import eu.weblibre.flutter_mozilla_components.pigeons.GeckoEngineSettings
 import eu.weblibre.flutter_mozilla_components.pigeons.GeckoGestureEvents
 import eu.weblibre.flutter_mozilla_components.pigeons.GeckoHistoryEvents
+import eu.weblibre.flutter_mozilla_components.pigeons.GeckoPushEvents
 import eu.weblibre.flutter_mozilla_components.pigeons.GeckoSelectionActionEvents
 import eu.weblibre.flutter_mozilla_components.pigeons.GeckoStateEvents
 import eu.weblibre.flutter_mozilla_components.pigeons.GeckoSuggestionEvents
@@ -33,11 +34,13 @@ import eu.weblibre.flutter_mozilla_components.api.GeckoViewportApiImpl
 import eu.weblibre.flutter_mozilla_components.api.GeckoEngineSettingsApiImpl
 import eu.weblibre.flutter_mozilla_components.feature.DefaultSelectionActionDelegate
 import eu.weblibre.flutter_mozilla_components.feature.GeckoBookmarksExtensionBridge
+import eu.weblibre.flutter_mozilla_components.push.Push
 import kotlinx.coroutines.DelicateCoroutinesApi
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.GlobalScope
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeoutOrNull
 import mozilla.components.browser.storage.sync.GlobalPlacesDependencyProvider
 import mozilla.components.browser.session.storage.RecoverableBrowserState
 import mozilla.components.browser.state.action.RestoreCompleteAction
@@ -64,6 +67,8 @@ private const val DEFAULT_QUERY_PARAMETER_STRIPPING_STRIP_LIST =
 private const val UBLOCK_FILTER_LISTS_PREF = "browser.weblibre.uBO.filterLists"
 private const val EXCLUDED_HISTORY_CONTEXT_IDS_PREF =
     "browser.weblibre.excludedHistoryContextIds"
+private const val PROFILE_SWITCH_PERSIST_TIMEOUT_MS = 3000L
+private const val PROFILE_SWITCH_DETACH_TIMEOUT_MS = 2000L
 
 object GlobalComponents {
     private var _components: Components? = null
@@ -74,6 +79,30 @@ object GlobalComponents {
     val components: Components?
         get() = _components
 
+    internal val isExternalMode: Boolean
+        get() = currentMode == ComponentsMode.EXTERNAL
+
+    /** Resolve a live Push only when it belongs to the supplied profile context. */
+    fun pushForProfile(context: Context): Push? {
+        val profilePath = (context as? ProfileContext)?.relativePath ?: return null
+        val current = _components ?: return null
+        if (current.profileApplicationContext.relativePath != profilePath) return null
+        return current.existingPush?.takeUnless { it.isClosed }
+    }
+
+    fun resolveActiveProfileContext(context: Context): ProfileContext? =
+        runCatching { ActiveProfile.resolveContext(context.applicationContext) }.getOrNull()
+
+    fun closePush() {
+        _components?.existingPush?.close()
+    }
+
+    fun tearDown() {
+        _components?.existingPush?.close()
+        _components = null
+        currentMode = null
+    }
+
     enum class ComponentsMode {
         FULL,
         EXTERNAL,
@@ -116,6 +145,11 @@ object GlobalComponents {
     // container contextIds but skips Dart relation emits.
     var historyEvents: GeckoHistoryEvents? = null
 
+    // Native -> Dart UnifiedPush registration lifecycle. Null when push events
+    // arrive with no Flutter engine attached (the UnifiedPushReceiver cold-start
+    // path), in which case failures are logged natively only.
+    var pushEvents: GeckoPushEvents? = null
+
     // Gecko contextIds of containers with hard exclude-from-history enabled.
     // Pushed from Dart; read by WebLibreHistoryDelegate to skip the Places
     // write for visits resolved to one of these containers.
@@ -334,6 +368,40 @@ object GlobalComponents {
             emptyList()
         }
 
+        previousComponents?.existingPush?.let { previousPush ->
+            if (!isSameProfile) {
+                val targetProfileId = File(applicationContext.relativePath).name
+                    .removePrefix(PwaConstants.PROFILE_DIR_PREFIX)
+                runBlocking {
+                    // Persist the switch while holding the profile lock so an
+                    // in-flight worker or receiver cannot straddle it. Bound only
+                    // the wait for exclusivity; once the atomic write starts it
+                    // must return a definitive result. Failure aborts setup before
+                    // B's components are created.
+                    check(
+                        previousPush.persistProfileSwitch(
+                            targetProfileId,
+                            PROFILE_SWITCH_PERSIST_TIMEOUT_MS,
+                        ),
+                    ) { "Timed out waiting to persist profile switch to $targetProfileId" }
+                    // Detaching the now-inactive profile's transport is best-effort
+                    // cleanup; bound it so a slow distributor cannot stall setup.
+                    runCatching {
+                        withTimeoutOrNull(PROFILE_SWITCH_DETACH_TIMEOUT_MS) {
+                            previousPush.detachTransportForSwitch()
+                        } ?: Logger.warn("Timed out detaching push transport during switch")
+                    }.onFailure {
+                        Logger.warn("Failed to detach push transport during switch", it)
+                    }
+                }
+                // Closing may need the same dispatcher as a timed-out detach.
+                // Mark it closed now, but drain old-profile resources off-main.
+                previousPush.closeDeferred()
+            } else {
+                previousPush.close()
+            }
+        }
+
         val newComponents = Components(
             applicationContext,
             flutterEvents,

packages/flutter_mozilla_components/android/src/main/kotlin/eu/weblibre/flutter_mozilla_components/ProfileContext.kt

@@ -12,6 +12,9 @@ import java.io.File
 class ProfileContext(private val base: Context, val relativePath: String) :
     ContextWrapper(base) {
 
+    internal val rootApplicationContext: Context
+        get() = base.applicationContext
+
     private val subfolderRoot =
         File(base.filesDir, relativePath) // /data/user/0/com.app/profiles/default
 

packages/flutter_mozilla_components/android/src/main/kotlin/eu/weblibre/flutter_mozilla_components/api/GeckoBrowserApiImpl.kt

@@ -47,6 +47,8 @@ import eu.weblibre.flutter_mozilla_components.pigeons.GeckoTrackingProtectionApi
 import eu.weblibre.flutter_mozilla_components.pigeons.GeckoLogging
 import eu.weblibre.flutter_mozilla_components.pigeons.GeckoMlApi
 import eu.weblibre.flutter_mozilla_components.pigeons.GeckoPrefApi
+import eu.weblibre.flutter_mozilla_components.pigeons.GeckoPushApi
+import eu.weblibre.flutter_mozilla_components.pigeons.GeckoPushEvents
 import eu.weblibre.flutter_mozilla_components.pigeons.GeckoPwaApi
 import eu.weblibre.flutter_mozilla_components.pigeons.GeckoSelectionActionController
 import eu.weblibre.flutter_mozilla_components.pigeons.GeckoHistoryEvents
@@ -132,6 +134,7 @@ class GeckoBrowserApiImpl : GeckoBrowserApi {
 
     private var activity: Activity? = null
     private var isPlatformViewRegistered = false
+    private var pushApi: GeckoPushApiImpl? = null
 
     private lateinit var _flutterPluginBinding: FlutterPlugin.FlutterPluginBinding
     private lateinit var _flutterEvents: GeckoStateEvents
@@ -167,6 +170,11 @@ class GeckoBrowserApiImpl : GeckoBrowserApi {
         this.activity = activity
     }
 
+    fun disposePushApi() {
+        pushApi?.dispose()
+        pushApi = null
+    }
+
     fun detachActivity() {
         this.activity = null
     }
@@ -272,6 +280,10 @@ class GeckoBrowserApiImpl : GeckoBrowserApi {
         GlobalComponents.historyEvents =
             GeckoHistoryEvents(_flutterPluginBinding.binaryMessenger)
 
+        // Also set before GlobalComponents.setUp, which calls push.initialize() and can therefore
+        // surface a registration failure before this sink would otherwise exist.
+        GlobalComponents.pushEvents = GeckoPushEvents(_flutterPluginBinding.binaryMessenger)
+
         GlobalComponents.setUp(
             profileApplicationContext,
             _flutterEvents,
@@ -362,6 +374,15 @@ class GeckoBrowserApiImpl : GeckoBrowserApi {
             GeckoGestureApiImpl()
         )
 
+        // UnifiedPush distributor management. The event sink was installed above, before
+        // GlobalComponents.setUp initialized push.
+        pushApi?.dispose()
+        pushApi = GeckoPushApiImpl()
+        GeckoPushApi.setUp(
+            _flutterPluginBinding.binaryMessenger,
+            pushApi
+        )
+
         ReaderViewEvents.setUp(
             _flutterPluginBinding.binaryMessenger,
             components.events.readerViewEvents
@@ -496,23 +517,6 @@ class GeckoBrowserApiImpl : GeckoBrowserApi {
         currentActivity.startActivity(intent)
     }
 
-    override fun pickUnifiedPushDistributor(callback: (Result<Boolean>) -> Unit) {
-        val currentActivity = activity
-        if (currentActivity == null) {
-            callback(Result.success(false))
-            return
-        }
-
-        runCatching {
-            components.push.pickDistributor(currentActivity) { success ->
-                callback(Result.success(success))
-            }
-        }.onFailure { error ->
-            logger.error("$TAG: Failed to pick UnifiedPush distributor", error)
-            callback(Result.failure(error))
-        }
-    }
-
     override fun shutdown() {
         logger.debug("$TAG: Shutting down GeckoView engine")
 
@@ -538,6 +542,8 @@ class GeckoBrowserApiImpl : GeckoBrowserApi {
         // 2. Stop component-level services
         try {
             GlobalComponents.stopPrivateTabsNotificationFeature()
+            disposePushApi()
+            GlobalComponents.closePush()
 
             GlobalComponents.components?.let { components ->
                 // Stop the FxA web channel feature
@@ -555,6 +561,8 @@ class GeckoBrowserApiImpl : GeckoBrowserApi {
             EngineProvider.shutdown()
         } catch (e: Exception) {
             logger.error("$TAG: Error shutting down GeckoRuntime", e)
+        } finally {
+            GlobalComponents.tearDown()
         }
 
         isGeckoInitialized = false

packages/flutter_mozilla_components/android/src/main/kotlin/eu/weblibre/flutter_mozilla_components/api/GeckoPushApiImpl.kt

@@ -0,0 +1,82 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
+
+package eu.weblibre.flutter_mozilla_components.api
+
+import eu.weblibre.flutter_mozilla_components.GlobalComponents
+import eu.weblibre.flutter_mozilla_components.pigeons.GeckoPushApi
+import eu.weblibre.flutter_mozilla_components.pigeons.PushStatus
+import eu.weblibre.flutter_mozilla_components.pigeons.PushSubscription
+import eu.weblibre.flutter_mozilla_components.push.toPigeon
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.withContext
+
+/** UnifiedPush distributor management for the settings UI. */
+class GeckoPushApiImpl : GeckoPushApi {
+    private val coroutineScope = CoroutineScope(Dispatchers.Main + SupervisorJob())
+
+    private val push
+        get() = requireNotNull(GlobalComponents.components) { "Components not initialized" }.push
+
+    override fun getPushStatus(callback: (Result<PushStatus>) -> Unit) {
+        respond(callback) {
+            withContext(Dispatchers.IO) { push.status() }.toPigeon()
+        }
+    }
+
+    override fun setDistributor(packageName: String, callback: (Result<Unit>) -> Unit) {
+        respond(callback) {
+            withContext(Dispatchers.IO) { push.setDistributor(packageName) }
+        }
+    }
+
+    override fun removeDistributor(callback: (Result<Unit>) -> Unit) {
+        respond(callback) {
+            withContext(Dispatchers.IO) { push.removeDistributor() }
+        }
+    }
+
+    override fun renewRegistration(callback: (Result<Unit>) -> Unit) {
+        respond(callback) {
+            withContext(Dispatchers.IO) { push.renewRegistration() }
+        }
+    }
+
+    override fun suspendForProfileSwitch(targetProfileId: String, callback: (Result<Unit>) -> Unit) {
+        respond(callback) {
+            withContext(Dispatchers.IO) { push.suspendForProfileSwitch(targetProfileId) }
+        }
+    }
+
+    override fun getSubscriptions(callback: (Result<List<PushSubscription>>) -> Unit) {
+        respond(callback) {
+            withContext(Dispatchers.IO) {
+                push.subscriptions().map {
+                    PushSubscription(scope = it.scope, hasEndpoint = it.hasEndpoint)
+                }
+            }
+        }
+    }
+
+    private fun <T> respond(callback: (Result<T>) -> Unit, block: suspend () -> T) {
+        coroutineScope.launch {
+            try {
+                callback(Result.success(block()))
+            } catch (error: CancellationException) {
+                throw error
+            } catch (error: Throwable) {
+                callback(Result.failure(error))
+            }
+        }
+    }
+
+    fun dispose() {
+        coroutineScope.cancel()
+    }
+}

packages/flutter_mozilla_components/android/src/main/kotlin/eu/weblibre/flutter_mozilla_components/components/Core.kt

@@ -33,6 +33,7 @@ import eu.weblibre.flutter_mozilla_components.middleware.SandboxCaptureMiddlewar
 import eu.weblibre.flutter_mozilla_components.middleware.SaveToPDFMiddleware
 import eu.weblibre.flutter_mozilla_components.pigeons.BrowserExtensionEvents
 import eu.weblibre.flutter_mozilla_components.pigeons.GeckoStateEvents
+import eu.weblibre.flutter_mozilla_components.push.WebNotificationDrainCoordinator
 import kotlinx.coroutines.FlowPreview
 import mozilla.components.browser.engine.gecko.permission.GeckoSitePermissionsStorage
 import mozilla.components.browser.engine.gecko.util.EngineDownloadDelegate
@@ -225,6 +226,11 @@ class Core(
         HistoryMetadataService(storage = historyStorage)
     }
 
+    // Wraps the WebNotificationFeature delegate so headless push deliveries can
+    // wait for the service worker to actually post its notification before the
+    // process loses foreground priority. Installed when [store] is created.
+    val webNotificationDrainCoordinator = WebNotificationDrainCoordinator()
+
     @OptIn(FlowPreview::class)
     val store by lazy {
         BrowserStore(
@@ -282,7 +288,11 @@ class Core(
 
             icons.install(engine, this)
 
-            WebNotificationFeature(
+            // WebNotificationFeature self-registers as the engine's notification
+            // delegate in its init; immediately wrap it with the drain
+            // coordinator so headless deliveries observe onShowNotification while
+            // notifications still display exactly as before.
+            val webNotificationFeature = WebNotificationFeature(
                 context,
                 engine,
                 icons,
@@ -291,6 +301,8 @@ class Core(
                 NotificationActivity::class.java,
                 notificationsDelegate = components.notificationsDelegate,
             )
+            webNotificationDrainCoordinator.delegate = webNotificationFeature
+            engine.registerWebNotificationDelegate(webNotificationDrainCoordinator)
 
             MediaSessionFeature(context, MediaSessionService::class.java, this).start()
         }

packages/flutter_mozilla_components/android/src/main/kotlin/eu/weblibre/flutter_mozilla_components/components/Push.kt

@@ -1,53 +0,0 @@
-/* This Source Code Form is subject to the terms of the Mozilla Public
- * License, v. 2.0. If a copy of the MPL was not distributed with this
- * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
-
-package eu.weblibre.flutter_mozilla_components.components
-
-import android.app.Activity
-import eu.weblibre.flutter_mozilla_components.Components
-import eu.weblibre.flutter_mozilla_components.push.WebPushEngineIntegration
-import java.util.concurrent.atomic.AtomicBoolean
-import org.ironfoxoss.unifiedpush.UnifiedPushFeature
-import org.unifiedpush.android.connector.UnifiedPush
-
-/**
- * Component group for web push services backed by UnifiedPush.
- */
-class Push(
-    private val components: Components,
-) {
-    private val initialized = AtomicBoolean(false)
-
-    private val feature by lazy {
-        UnifiedPushFeature(
-            context = components.profileApplicationContext,
-            disableRateLimit = true,
-        )
-    }
-
-    private val webPushEngineIntegration by lazy {
-        WebPushEngineIntegration(components.core.engine, feature)
-    }
-
-    fun initialize() {
-        if (!initialized.compareAndSet(false, true)) {
-            return
-        }
-
-        // Ensure the store-side WebNotificationFeature is installed before push events arrive.
-        components.core.store
-        webPushEngineIntegration.start()
-        feature.initialize()
-    }
-
-    fun pickDistributor(activity: Activity, callback: (Boolean) -> Unit) {
-        initialize()
-        UnifiedPush.tryPickDistributor(activity) { success ->
-            if (success) {
-                feature.renewRegistration()
-            }
-            callback(success)
-        }
-    }
-}

packages/flutter_mozilla_components/android/src/main/kotlin/eu/weblibre/flutter_mozilla_components/push/Push.kt

@@ -0,0 +1,473 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
+
+package eu.weblibre.flutter_mozilla_components.push
+
+import android.content.Context
+import android.content.pm.PackageManager
+import android.util.Log
+import androidx.core.app.NotificationManagerCompat
+import eu.weblibre.flutter_mozilla_components.Components
+import eu.weblibre.flutter_mozilla_components.GlobalComponents
+import eu.weblibre.flutter_mozilla_components.ActiveProfile
+import eu.weblibre.flutter_mozilla_components.ext.EventSequence
+import java.util.concurrent.Executors
+import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.atomic.AtomicInteger
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.ExecutorCoroutineDispatcher
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.asCoroutineDispatcher
+import kotlinx.coroutines.async
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.cancelAndJoin
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeoutOrNull
+import kotlinx.coroutines.withContext
+import org.ironfoxoss.unifiedpush.PushError
+import org.ironfoxoss.unifiedpush.SubscriptionsDB
+import org.ironfoxoss.unifiedpush.UnifiedPushFeature
+import org.ironfoxoss.unifiedpush.UnifiedPushNotification
+import org.unifiedpush.android.connector.UnifiedPush
+import org.unifiedpush.android.connector.data.PushEndpoint
+import org.mozilla.gecko.GeckoThread
+import mozilla.components.support.ktx.kotlin.getOrigin
+
+private const val START_WAITING = 0
+private const val STARTED = 1
+private const val START_TIMED_OUT = 2
+
+/**
+ * Bounds only the wait for [operation] to call its start gate. Once started,
+ * the operation is allowed to finish so a completed side effect cannot be
+ * reported as a timeout.
+ */
+internal suspend fun runWithStartTimeout(
+    timeoutMillis: Long,
+    operation: suspend (tryStart: () -> Boolean) -> Unit,
+): Boolean = coroutineScope {
+    require(timeoutMillis > 0) { "Timeout must be positive" }
+
+    val state = AtomicInteger(START_WAITING)
+    val started = CompletableDeferred<Unit>()
+    val operationJob = async {
+        operation {
+            if (!state.compareAndSet(START_WAITING, STARTED)) {
+                false
+            } else {
+                started.complete(Unit)
+                true
+            }
+        }
+    }
+
+    val startedBeforeTimeout = withTimeoutOrNull(timeoutMillis) {
+        started.await()
+        true
+    } == true
+
+    if (!startedBeforeTimeout && state.compareAndSet(START_WAITING, START_TIMED_OUT)) {
+        operationJob.cancelAndJoin()
+        false
+    } else {
+        operationJob.await()
+        true
+    }
+}
+
+/** Lifecycle state of the selected UnifiedPush distributor. */
+enum class DistributorStatus {
+    NONE_AVAILABLE,
+    NOT_SELECTED,
+    PENDING,
+    READY,
+    UNAVAILABLE,
+}
+
+data class DistributorInfo(val packageName: String, val label: String?)
+
+data class PushStatusSnapshot(
+    val status: DistributorStatus,
+    val current: DistributorInfo?,
+    val available: List<DistributorInfo>,
+    val lastError: String?,
+)
+
+data class PushSubscriptionInfo(val scope: String, val hasEndpoint: Boolean)
+
+/** Profile-scoped UnifiedPush state and Gecko web-push integration. */
+class Push(
+    private val components: Components,
+) : AutoCloseable {
+    private val initialized = AtomicBoolean(false)
+    private val closed = AtomicBoolean(false)
+    internal val isClosed: Boolean
+        get() = closed.get()
+    private val dispatcher: ExecutorCoroutineDispatcher =
+        Executors.newSingleThreadExecutor { runnable ->
+            Thread(runnable, "WebLibrePush-${components.profileApplicationContext.relativePath.hashCode()}")
+        }.asCoroutineDispatcher()
+    private val eventScope = CoroutineScope(dispatcher + SupervisorJob())
+
+    private val context: Context
+        get() = components.profileApplicationContext
+
+    private val prefs = PushProfileState.prefs(context)
+    private val subscriptionsDb = SubscriptionsDB(context)
+
+    val feature = UnifiedPushFeature(
+        context = context,
+        coroutineContext = dispatcher,
+        db = subscriptionsDb,
+    )
+
+    private val webPushEngineIntegration =
+        WebPushEngineIntegration(components.core.engine, feature)
+
+    fun initialize() {
+        check(!closed.get()) { "Push is closed" }
+        if (!initialized.compareAndSet(false, true)) return
+
+        restoreRememberedDistributor()
+        components.core.store
+        webPushEngineIntegration.start()
+        feature.initialize()
+        eventScope.launch {
+            while (!PushMessageScheduler.recover(components.profileApplicationContext)) {
+                delay(RECOVERY_RETRY_DELAY_MS)
+            }
+            notifyIfDistributorMissing()
+        }
+    }
+
+    fun status(): PushStatusSnapshot {
+        val available = UnifiedPush.getDistributors(context).map { it.toDistributorInfo() }
+        val acknowledged = UnifiedPush.getAckDistributor(context)
+        val saved = UnifiedPush.getSavedDistributor(context)
+        val remembered = rememberedDistributor()
+        val status = when {
+            acknowledged != null -> DistributorStatus.READY
+            saved != null -> DistributorStatus.PENDING
+            remembered != null && available.none { it.packageName == remembered } ->
+                DistributorStatus.UNAVAILABLE
+            available.isEmpty() -> DistributorStatus.NONE_AVAILABLE
+            else -> DistributorStatus.NOT_SELECTED
+        }
+
+        return PushStatusSnapshot(
+            status = status,
+            current = (acknowledged ?: saved ?: remembered)?.toDistributorInfo(),
+            available = available,
+            lastError = PushProfileState.lastError(context),
+        )
+    }
+
+    suspend fun setDistributor(packageName: String) = UnifiedPushReceiver.runExclusive {
+        withContext(dispatcher) {
+            check(!closed.get()) { "Push is closed" }
+            require(UnifiedPush.getDistributors(context).contains(packageName)) {
+                "UnifiedPush distributor is not installed: $packageName"
+            }
+
+            val current = UnifiedPush.getSavedDistributor(context) ?: rememberedDistributor()
+            if (current != null && current != packageName) {
+                removeTransportRegistrationsAndEndpoints()
+            }
+            UnifiedPush.saveDistributor(context, packageName)
+            rememberDistributor(packageName)
+            PushProfileState.clearError(context)
+            cancelMissingDistributorNotification()
+            feature.renewRegistration()
+        }
+    }
+
+    suspend fun removeDistributor() = UnifiedPushReceiver.runExclusive {
+        withContext(dispatcher) {
+            check(!closed.get()) { "Push is closed" }
+            removeTransportRegistrationsAndEndpoints()
+            prefs.edit().remove(PushProfileState.KEY_SELECTED_DISTRIBUTOR).commit()
+            PushProfileState.clearError(context)
+            cancelMissingDistributorNotification()
+        }
+    }
+
+    suspend fun renewRegistration() = UnifiedPushReceiver.runExclusive {
+        withContext(dispatcher) {
+            check(!closed.get()) { "Push is closed" }
+            restoreRememberedDistributor()
+            feature.renewRegistration()
+        }
+    }
+
+    suspend fun subscriptions(): List<PushSubscriptionInfo> = withContext(dispatcher) {
+        subscriptionsDb.listSubscriptions().map {
+            PushSubscriptionInfo(scope = it.scope, hasEndpoint = it.endpoint != null)
+        }
+    }
+
+    suspend fun onNewEndpoint(scope: String, endpoint: PushEndpoint) = withContext(dispatcher) {
+        feature.onNewEndpoint(scope, endpoint)
+        if (endpoint.pubKeySet != null) PushProfileState.clearError(context, scope)
+    }
+
+    suspend fun invalidateEndpoint(scope: String) = withContext(dispatcher) {
+        subscriptionsDb.removeEndpoint(scope)
+        PushProfileState.clearError(context, scope)
+        webPushEngineIntegration.invalidateEndpoint(scope)
+    }
+
+    suspend fun onUnregistered(scope: String) = invalidateEndpoint(scope)
+
+    suspend fun recordRegistrationError(scope: String, error: PushError) = withContext(dispatcher) {
+        PushProfileState.recordError(
+            context,
+            scope,
+            PushProfileState.errorType(error),
+            error.message,
+        )
+    }
+
+    suspend fun recordTemporaryUnavailable(scope: String) = withContext(dispatcher) {
+        PushProfileState.recordTemporaryUnavailable(context, scope)
+    }
+
+    suspend fun deliverMessage(scope: String, payload: ByteArray) {
+        val external = GlobalComponents.isExternalMode
+        val deliver: suspend () -> Unit = {
+            withContext(Dispatchers.Main.immediate) {
+                check(!closed.get()) { "Push is closed" }
+                check(GeckoThread.isStateAtLeast(GeckoThread.State.RUNNING)) {
+                    "Gecko is not running"
+                }
+                webPushEngineIntegration.deliverMessage(scope, payload)
+            }
+        }
+
+        if (!external) {
+            deliver()
+            return
+        }
+
+        // Headless: the push message is decrypted and permitted, but GeckoView
+        // will not run the service worker's push handler without a live browsing
+        // context — with no open session the ServiceWorkerManager never dispatches
+        // the event (opening a tab is what makes it fire). Create a throwaway
+        // session for the duration of delivery so Gecko has a window to run the
+        // worker in, then tear it down.
+        val origin = runCatching { scope.getOrigin() }.getOrNull()
+        val session = withContext(Dispatchers.Main.immediate) {
+            components.core.engine.createSession().also { it.loadUrl("about:blank") }
+        }
+        try {
+            // Give the browsing context time to come up before handing off the push.
+            delay(HEADLESS_SESSION_WARMUP_MS)
+            // The push handoff returns no completion signal, so keep this delivery
+            // alive until the service worker actually posts its notification,
+            // bounded by a timeout.
+            components.core.webNotificationDrainCoordinator.drainWhileDelivering(
+                origin = origin,
+                timeoutMillis = HEADLESS_DELIVERY_DRAIN_TIMEOUT_MS,
+                graceMillis = HEADLESS_DELIVERY_POST_GRACE_MS,
+                deliver = deliver,
+            )
+        } finally {
+            withContext(Dispatchers.Main.immediate) { session.close() }
+        }
+    }
+
+    /**
+     * Persist the profile switch while holding the profile lock, so an in-flight
+     * delivery worker (which holds the same lock for the duration of a delivery)
+     * cannot straddle the switch and deliver this profile's message after disk
+     * state has moved on. Throws on failure so the caller can abort rather than
+     * proceed with an inconsistent on-disk profile.
+     */
+    suspend fun persistProfileSwitch(targetProfileId: String) {
+        check(!closed.get()) { "Push is closed" }
+        persistProfileSwitch(targetProfileId) { true }
+    }
+
+    /**
+     * Persist the switch if profile and receiver exclusivity can be obtained
+     * within [startTimeoutMillis]. The timeout stops applying once the atomic
+     * file write starts.
+     */
+    suspend fun persistProfileSwitch(
+        targetProfileId: String,
+        startTimeoutMillis: Long,
+    ): Boolean {
+        check(!closed.get()) { "Push is closed" }
+        return runWithStartTimeout(startTimeoutMillis) { tryStart ->
+            persistProfileSwitch(targetProfileId, tryStart)
+        }
+    }
+
+    private suspend fun persistProfileSwitch(
+        targetProfileId: String,
+        tryStart: () -> Boolean,
+    ) {
+        ActiveProfile.withProfileLock {
+            UnifiedPushReceiver.runExclusive {
+                if (tryStart()) {
+                    ActiveProfile.switchTo(
+                        components.profileApplicationContext.rootApplicationContext,
+                        targetProfileId,
+                    )
+                }
+            }
+        }
+    }
+
+    /**
+     * Detach the now-inactive profile's push transport. Best-effort: a stale
+     * registration is harmless and is cleaned up when that profile next becomes
+     * active. Subscriptions and the remembered distributor are preserved.
+     */
+    suspend fun detachTransportForSwitch() {
+        ActiveProfile.withProfileLock {
+            UnifiedPushReceiver.runExclusive {
+                withContext(dispatcher) {
+                    if (!closed.get()) {
+                        removeTransportRegistrationsAndEndpoints(notifyGecko = false)
+                    }
+                }
+            }
+        }
+    }
+
+    /** Persist the switch (mandatory), then best-effort detach the old transport. */
+    suspend fun suspendForProfileSwitch(targetProfileId: String) {
+        persistProfileSwitch(targetProfileId)
+        runCatching { detachTransportForSwitch() }
+            .onFailure { error ->
+                Log.w(TAG, "Failed to detach push transport during profile switch", error)
+            }
+    }
+
+    fun emitStatusChanged() {
+        if (closed.get() || GlobalComponents.pushEvents == null) return
+        eventScope.launch {
+            val snapshot = runCatching { status().toPigeon() }.getOrNull() ?: return@launch
+            val sequence = EventSequence.next()
+            withContext(kotlinx.coroutines.Dispatchers.Main) {
+                GlobalComponents.pushEvents?.onPushStatusChanged(sequence, snapshot) { }
+            }
+        }
+    }
+
+    override fun close() {
+        if (!beginClose()) return
+        try {
+            runBlocking { finishClose() }
+        } finally {
+            dispatcher.close()
+        }
+    }
+
+    /** Mark closed immediately and drain profile resources without blocking the caller. */
+    internal fun closeDeferred() {
+        if (!beginClose()) return
+        CoroutineScope(SupervisorJob() + Dispatchers.IO).launch {
+            try {
+                finishClose()
+            } catch (error: Throwable) {
+                Log.w(TAG, "Failed to finish deferred push cleanup", error)
+            } finally {
+                dispatcher.close()
+            }
+        }
+    }
+
+    private fun beginClose(): Boolean {
+        if (!closed.compareAndSet(false, true)) return false
+        eventScope.cancel()
+        if (initialized.get()) {
+            webPushEngineIntegration.close()
+        }
+        return true
+    }
+
+    private suspend fun finishClose() {
+        if (initialized.get()) {
+            val drained = CompletableDeferred<Unit>()
+            feature.withCoroutine { drained.complete(Unit) }
+            withTimeoutOrNull(FEATURE_DRAIN_TIMEOUT_MS) { drained.await() }
+        }
+        withContext(dispatcher) {
+            subscriptionsDb.close()
+        }
+    }
+
+    private fun restoreRememberedDistributor() {
+        if (UnifiedPush.getSavedDistributor(context) != null) return
+        val remembered = rememberedDistributor() ?: return
+        if (UnifiedPush.getDistributors(context).contains(remembered)) {
+            UnifiedPush.saveDistributor(context, remembered)
+        }
+    }
+
+    private suspend fun removeTransportRegistrationsAndEndpoints(notifyGecko: Boolean = true) {
+        UnifiedPush.removeDistributor(context)
+        subscriptionsDb.listSubscriptions().forEach {
+            subscriptionsDb.removeEndpoint(it.scope)
+            if (notifyGecko) webPushEngineIntegration.invalidateEndpoint(it.scope)
+        }
+    }
+
+    private fun rememberedDistributor(): String? =
+        prefs.getString(PushProfileState.KEY_SELECTED_DISTRIBUTOR, null)
+
+    private fun rememberDistributor(packageName: String) {
+        prefs.edit().putString(PushProfileState.KEY_SELECTED_DISTRIBUTOR, packageName).commit()
+    }
+
+    private fun notifyIfDistributorMissing() {
+        if (status().status != DistributorStatus.UNAVAILABLE) return
+        val notificationManager = NotificationManagerCompat.from(context)
+        if (!notificationManager.areNotificationsEnabled()) return
+        try {
+            notificationManager.notify(
+                UnifiedPushNotification.getNotificationId(context),
+                UnifiedPushNotification.createMissingServiceNotification(context),
+            )
+        } catch (_: SecurityException) {
+            // The settings status remains available when POST_NOTIFICATIONS is denied.
+        }
+    }
+
+    private fun cancelMissingDistributorNotification() {
+        NotificationManagerCompat.from(context)
+            .cancel(UnifiedPushNotification.getNotificationId(context))
+    }
+
+    private fun String.toDistributorInfo(): DistributorInfo =
+        DistributorInfo(packageName = this, label = resolveLabel(this))
+
+    private fun resolveLabel(packageName: String): String? = try {
+        val packageManager = context.packageManager
+        packageManager.getApplicationInfo(packageName, 0).loadLabel(packageManager).toString()
+    } catch (_: PackageManager.NameNotFoundException) {
+        null
+    }
+
+    companion object {
+        private const val TAG = "Push"
+        private const val FEATURE_DRAIN_TIMEOUT_MS = 5000L
+        // Upper bound on how long a headless delivery keeps the worker alive
+        // waiting for the service worker to post its notification.
+        private const val HEADLESS_DELIVERY_DRAIN_TIMEOUT_MS = 25000L
+        // Extra time after onShowNotification fires so the delegate's async
+        // notify can land before the process loses foreground priority.
+        private const val HEADLESS_DELIVERY_POST_GRACE_MS = 1500L
+        // Time for the throwaway delivery session's browsing context to come up
+        // before the push is handed off.
+        private const val HEADLESS_SESSION_WARMUP_MS = 1500L
+        private const val RECOVERY_RETRY_DELAY_MS = 30000L
+    }
+}

packages/flutter_mozilla_components/android/src/main/kotlin/eu/weblibre/flutter_mozilla_components/push/PushMessageScheduler.kt

@@ -0,0 +1,96 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
+
+package eu.weblibre.flutter_mozilla_components.push
+
+import android.util.Log
+import androidx.work.BackoffPolicy
+import androidx.work.Data
+import androidx.work.ExistingWorkPolicy
+import androidx.work.OneTimeWorkRequestBuilder
+import androidx.work.OutOfQuotaPolicy
+import androidx.work.WorkManager
+import eu.weblibre.flutter_mozilla_components.ProfileContext
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+import mozilla.components.support.ktx.android.content.runOnlyInMainProcess
+
+object PushMessageScheduler {
+    fun enqueue(context: ProfileContext, messageId: String) {
+        val request = OneTimeWorkRequestBuilder<PushMessageWorker>()
+            .setExpedited(OutOfQuotaPolicy.RUN_AS_NON_EXPEDITED_WORK_REQUEST)
+            .setBackoffCriteria(
+                BackoffPolicy.EXPONENTIAL,
+                MIN_BACKOFF_SECONDS,
+                TimeUnit.SECONDS,
+            )
+            .setInputData(
+                Data.Builder()
+                    .putString(PushMessageWorker.KEY_PROFILE_PATH, context.relativePath)
+                    .putString(PushMessageWorker.KEY_MESSAGE_ID, messageId)
+                    .build(),
+            )
+            .build()
+        val operation = WorkManager.getInstance(context)
+            .enqueueUniqueWork(workName(context.relativePath, messageId), ExistingWorkPolicy.KEEP, request)
+        operation.result.addListener(
+            {
+                runCatching { operation.result.get() }.onFailure { error ->
+                    Log.e(TAG, "Unable to enqueue push message $messageId", error)
+                    recoverLater(context)
+                }
+            },
+            recoveryExecutor,
+        )
+    }
+
+    fun recover(context: ProfileContext): Boolean {
+        val store = PushMessageStore(context)
+        return store.ids().map { messageId ->
+            runCatching { enqueue(context, messageId) }
+                .onFailure { error ->
+                    Log.e(TAG, "Unable to recover queued push message $messageId", error)
+                }
+                .isSuccess
+        }.all { it }
+    }
+
+    fun recoverLater(context: ProfileContext) {
+        context.runOnlyInMainProcess {
+            if (!recoveringProfiles.add(context.relativePath)) return@runOnlyInMainProcess
+            scheduleRecovery(context, 0)
+        }
+    }
+
+    private fun scheduleRecovery(context: ProfileContext, delaySeconds: Long) {
+        recoveryExecutor.schedule(
+            {
+                val recovered = runCatching { recover(context) }
+                    .onFailure { error ->
+                        Log.e(TAG, "Queued push recovery failed for ${context.relativePath}", error)
+                    }
+                    .getOrDefault(false)
+                if (recovered) {
+                    recoveringProfiles.remove(context.relativePath)
+                } else {
+                    scheduleRecovery(context, RECOVERY_RETRY_SECONDS)
+                }
+            },
+            delaySeconds,
+            TimeUnit.SECONDS,
+        )
+    }
+
+    private fun workName(profilePath: String, messageId: String): String =
+        "push-message-$profilePath-$messageId"
+
+    private const val MIN_BACKOFF_SECONDS = 10L
+    private const val RECOVERY_RETRY_SECONDS = 30L
+    private const val TAG = "PushMessageScheduler"
+    private val recoveringProfiles = ConcurrentHashMap.newKeySet<String>()
+    private val recoveryExecutor = Executors.newSingleThreadScheduledExecutor { runnable ->
+        Thread(runnable, "PushMessageRecovery").apply { isDaemon = true }
+    }
+}

packages/flutter_mozilla_components/android/src/main/kotlin/eu/weblibre/flutter_mozilla_components/push/PushMessageStore.kt

@@ -0,0 +1,191 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
+
+package eu.weblibre.flutter_mozilla_components.push
+
+import android.content.Context
+import android.system.Os
+import android.system.OsConstants
+import java.io.DataInputStream
+import java.io.DataOutputStream
+import java.io.File
+import java.io.FileDescriptor
+import java.io.FileNotFoundException
+import java.io.FileInputStream
+import java.io.FileOutputStream
+import java.io.IOException
+import java.util.UUID
+
+data class StoredPushMessage(
+    val id: String,
+    val scope: String,
+    val payload: ByteArray,
+)
+
+internal class CorruptPushMessageException(
+    message: String,
+    cause: Throwable? = null,
+) : IOException(message, cause)
+
+/** Profile-scoped, crash-safe queue storage for decrypted push payloads. */
+class PushMessageStore internal constructor(
+    private val directory: File,
+) {
+    constructor(context: Context) : this(File(context.noBackupFilesDir, DIRECTORY_NAME))
+
+    @Synchronized
+    fun persist(scope: String, payload: ByteArray, id: String = UUID.randomUUID().toString()): StoredPushMessage {
+        require(id.matches(SAFE_ID)) { "Invalid push message id" }
+        require(payload.size <= MAX_PAYLOAD_BYTES) { "Push payload is too large" }
+        directory.mkdirs()
+        if (isCompleted(id)) return StoredPushMessage(id, scope, payload.copyOf())
+
+        val target = file(id)
+        val temporary = File(directory, ".$id.tmp")
+        try {
+            FileOutputStream(temporary).use { output ->
+                DataOutputStream(output).use { data ->
+                    data.writeInt(FORMAT_VERSION)
+                    data.writeUTF(scope)
+                    data.writeInt(payload.size)
+                    data.write(payload)
+                    data.flush()
+                    output.fd.sync()
+                }
+            }
+            check(temporary.renameTo(target)) { "Unable to persist push message" }
+            syncDirectory()
+        } finally {
+            temporary.delete()
+        }
+        return StoredPushMessage(id, scope, payload.copyOf())
+    }
+
+    @Synchronized
+    fun get(id: String): StoredPushMessage? {
+        val source = file(id)
+        if (!source.isFile) return null
+        if (isCompleted(id)) return null
+
+        try {
+            DataInputStream(FileInputStream(source)).use { data ->
+                if (data.readInt() != FORMAT_VERSION) {
+                    throw CorruptPushMessageException("Unsupported push message format")
+                }
+                val scope = data.readUTF()
+                val size = data.readInt()
+                if (size < 0 || size > MAX_PAYLOAD_BYTES) {
+                    throw CorruptPushMessageException("Invalid push payload size")
+                }
+                val payload = ByteArray(size)
+                data.readFully(payload)
+                return StoredPushMessage(id, scope, payload)
+            }
+        } catch (error: CorruptPushMessageException) {
+            throw error
+        } catch (error: FileNotFoundException) {
+            if (!source.exists()) return null
+            throw CorruptPushMessageException("Unable to open push message", error)
+        } catch (error: IOException) {
+            throw CorruptPushMessageException("Unable to read push message", error)
+        }
+    }
+
+    @Synchronized
+    fun ids(): List<String> {
+        if (!directory.isDirectory) return emptyList()
+        val files = directory.listFiles().orEmpty()
+        files.filter {
+            it.isFile &&
+                it.extension == COMPLETED_EXTENSION &&
+                it.nameWithoutExtension.matches(SAFE_ID)
+        }.forEach { isCompleted(it.nameWithoutExtension) }
+
+        return files
+            .filter {
+                it.isFile &&
+                    it.extension == FILE_EXTENSION &&
+                    it.nameWithoutExtension.matches(SAFE_ID)
+            }
+            .filterNot {
+                val id = it.nameWithoutExtension
+                isCompleted(id)
+            }
+            .map { it.nameWithoutExtension }
+    }
+
+    @Synchronized
+    fun complete(id: String): Boolean {
+        val source = file(id)
+        val completed = completedFile(id)
+        if (!source.exists()) {
+            return true
+        }
+
+        if (!completed.isFile) {
+            directory.mkdirs()
+            val temporary = File(directory, ".$id.$COMPLETED_EXTENSION.tmp")
+            try {
+                FileOutputStream(temporary).use { output ->
+                    output.write(COMPLETED_MARKER)
+                    output.flush()
+                    output.fd.sync()
+                }
+                if (!temporary.renameTo(completed)) return false
+                syncDirectory()
+            } finally {
+                temporary.delete()
+            }
+        }
+
+        source.delete()
+        return true
+    }
+
+    private fun file(id: String): File {
+        require(id.matches(SAFE_ID)) { "Invalid push message id" }
+        return File(directory, "$id.$FILE_EXTENSION")
+    }
+
+    private fun completedFile(id: String): File {
+        require(id.matches(SAFE_ID)) { "Invalid push message id" }
+        return File(directory, "$id.$COMPLETED_EXTENSION")
+    }
+
+    private fun isCompleted(id: String): Boolean {
+        val completed = completedFile(id)
+        if (!completed.isFile) return false
+
+        val source = file(id)
+        if (source.exists()) source.delete()
+        if (source.exists()) return true
+
+        val expired = System.currentTimeMillis() - completed.lastModified() >= COMPLETED_TTL_MS
+        if (expired && completed.delete()) return false
+        return completed.exists()
+    }
+
+    private fun syncDirectory() {
+        var descriptor: FileDescriptor? = null
+        try {
+            descriptor = Os.open(directory.path, OsConstants.O_RDONLY, 0)
+            Os.fsync(descriptor)
+        } catch (_: Exception) {
+            // File fsync remains the fallback on platforms that cannot fsync directories.
+        } finally {
+            descriptor?.let { runCatching { Os.close(it) } }
+        }
+    }
+
+    companion object {
+        private const val DIRECTORY_NAME = "queued_push_messages"
+        private const val FILE_EXTENSION = "push"
+        private const val COMPLETED_EXTENSION = "delivered"
+        private const val COMPLETED_MARKER = 1
+        private const val COMPLETED_TTL_MS = 7 * 24 * 60 * 60 * 1000L
+        private const val FORMAT_VERSION = 1
+        private const val MAX_PAYLOAD_BYTES = 16 * 1024 * 1024
+        private val SAFE_ID = Regex("[A-Za-z0-9_-]+")
+    }
+}

packages/flutter_mozilla_components/android/src/main/kotlin/eu/weblibre/flutter_mozilla_components/push/PushMessageWorker.kt

@@ -0,0 +1,114 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
+
+package eu.weblibre.flutter_mozilla_components.push
+
+import android.app.NotificationChannel
+import android.app.NotificationManager
+import android.content.Context
+import android.os.Build
+import android.util.Log
+import androidx.core.app.NotificationCompat
+import androidx.work.CoroutineWorker
+import androidx.work.ForegroundInfo
+import androidx.work.WorkerParameters
+import eu.weblibre.flutter_mozilla_components.ActiveProfile
+import eu.weblibre.flutter_mozilla_components.GlobalComponents
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.withContext
+
+class PushMessageWorker(
+    appContext: Context,
+    params: WorkerParameters,
+) : CoroutineWorker(appContext, params) {
+    override suspend fun getForegroundInfo(): ForegroundInfo {
+        if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.O) {
+            applicationContext.getSystemService(NotificationManager::class.java)
+                .createNotificationChannel(
+                    NotificationChannel(
+                        FOREGROUND_CHANNEL_ID,
+                        "Web notification delivery",
+                        NotificationManager.IMPORTANCE_LOW,
+                    ).apply {
+                        description = "Keeps web notification delivery active"
+                        setShowBadge(false)
+                    },
+                )
+        }
+
+        val appLabel = applicationContext.applicationInfo
+            .loadLabel(applicationContext.packageManager)
+        val notification = NotificationCompat.Builder(applicationContext, FOREGROUND_CHANNEL_ID)
+            .setSmallIcon(android.R.drawable.stat_notify_sync_noanim)
+            .setContentTitle(appLabel)
+            .setContentText("Delivering web notification")
+            .setCategory(NotificationCompat.CATEGORY_SERVICE)
+            .setPriority(NotificationCompat.PRIORITY_LOW)
+            .setLocalOnly(true)
+            .setOngoing(true)
+            .setSilent(true)
+            .setShowWhen(false)
+            .build()
+        val notificationId = (id.hashCode() and Int.MAX_VALUE).coerceAtLeast(1)
+        return ForegroundInfo(notificationId, notification)
+    }
+
+    override suspend fun doWork(): Result {
+        val queuedProfile = inputData.getString(KEY_PROFILE_PATH) ?: return Result.failure()
+        val messageId = inputData.getString(KEY_MESSAGE_ID) ?: return Result.failure()
+        return ActiveProfile.withProfileLock profile@{
+            val activeProfile = runCatching { ActiveProfile.resolveContext(applicationContext) }.getOrNull()
+                ?: return@profile Result.retry()
+            // Keep the durable record for recovery when this profile becomes active again.
+            if (activeProfile.relativePath != queuedProfile) return@profile Result.success()
+
+            val existing = GlobalComponents.components
+            if (existing != null && existing.profileApplicationContext.relativePath != queuedProfile) {
+                return@profile Result.success()
+            }
+            val initialized = existing != null || withContext(Dispatchers.Main.immediate) {
+                GlobalComponents.ensureExternalComponents(applicationContext)
+            }
+            if (!initialized) return@profile Result.retry()
+
+            val push = GlobalComponents.pushForProfile(activeProfile) ?: return@profile Result.retry()
+            val store = PushMessageStore(activeProfile)
+            val message = try {
+                store.get(messageId)
+            } catch (error: CorruptPushMessageException) {
+                Log.e(TAG, "Discarding corrupt queued push message $messageId", error)
+                if (!store.complete(messageId)) {
+                    Log.e(TAG, "Unable to mark corrupt push message $messageId as discarded")
+                }
+                return@profile Result.failure()
+            } ?: return@profile Result.success()
+
+            try {
+                push.deliverMessage(message.scope, message.payload)
+                if (!store.complete(message.id)) {
+                    Log.e(TAG, "Unable to mark delivered push message ${message.id} complete")
+                    return@profile Result.retry()
+                }
+                Result.success()
+            } catch (error: CancellationException) {
+                throw error
+            } catch (error: Throwable) {
+                Log.w(
+                    TAG,
+                    "Push delivery attempt ${runAttemptCount + 1} failed for $messageId",
+                    error,
+                )
+                Result.retry()
+            }
+        }
+    }
+
+    companion object {
+        const val KEY_PROFILE_PATH = "profilePath"
+        const val KEY_MESSAGE_ID = "messageId"
+        private const val FOREGROUND_CHANNEL_ID = "weblibre_push_delivery"
+        private const val TAG = "PushMessageWorker"
+    }
+}

packages/flutter_mozilla_components/android/src/main/kotlin/eu/weblibre/flutter_mozilla_components/push/PushPigeonMappers.kt

@@ -0,0 +1,27 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
+
+package eu.weblibre.flutter_mozilla_components.push
+
+import eu.weblibre.flutter_mozilla_components.pigeons.PushDistributor
+import eu.weblibre.flutter_mozilla_components.pigeons.PushDistributorStatus
+import eu.weblibre.flutter_mozilla_components.pigeons.PushStatus
+
+internal fun PushStatusSnapshot.toPigeon() = PushStatus(
+    status = status.toPigeon(),
+    current = current?.toPigeon(),
+    available = available.map { it.toPigeon() },
+    lastError = lastError,
+)
+
+internal fun DistributorInfo.toPigeon() =
+    PushDistributor(packageName = packageName, label = label)
+
+internal fun DistributorStatus.toPigeon() = when (this) {
+    DistributorStatus.NONE_AVAILABLE -> PushDistributorStatus.NONE_AVAILABLE
+    DistributorStatus.NOT_SELECTED -> PushDistributorStatus.NOT_SELECTED
+    DistributorStatus.PENDING -> PushDistributorStatus.PENDING
+    DistributorStatus.READY -> PushDistributorStatus.READY
+    DistributorStatus.UNAVAILABLE -> PushDistributorStatus.UNAVAILABLE
+}

packages/flutter_mozilla_components/android/src/main/kotlin/eu/weblibre/flutter_mozilla_components/push/PushProfileState.kt

@@ -0,0 +1,72 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
+
+package eu.weblibre.flutter_mozilla_components.push
+
+import android.content.Context
+import android.content.SharedPreferences
+import org.ironfoxoss.unifiedpush.PushError
+import org.ironfoxoss.unifiedpush.SubscriptionsDB
+import org.unifiedpush.android.connector.data.PushEndpoint
+
+internal object PushProfileState {
+    private const val PREFS_NAME = "weblibre_push"
+    const val KEY_SELECTED_DISTRIBUTOR = "selected_distributor"
+    private const val KEY_LAST_ERROR = "last_error"
+    private const val KEY_LAST_ERROR_SCOPE = "last_error_scope"
+    private const val KEY_LAST_ERROR_TYPE = "last_error_type"
+
+    fun lastError(context: Context): String? = prefs(context).getString(KEY_LAST_ERROR, null)
+
+    fun recordTemporaryUnavailable(context: Context, scope: String) {
+        recordError(
+            context,
+            scope,
+            "temporary_unavailable",
+            "Push service is temporarily unavailable",
+        )
+    }
+
+    fun recordError(context: Context, scope: String, type: String, message: String) {
+        prefs(context).edit()
+            .putString(KEY_LAST_ERROR, message)
+            .putString(KEY_LAST_ERROR_SCOPE, scope)
+            .putString(KEY_LAST_ERROR_TYPE, type)
+            .commit()
+    }
+
+    fun clearError(context: Context, scope: String? = null) {
+        val prefs = prefs(context)
+        if (scope != null && prefs.getString(KEY_LAST_ERROR_SCOPE, null) != scope) return
+        prefs.edit()
+            .remove(KEY_LAST_ERROR)
+            .remove(KEY_LAST_ERROR_SCOPE)
+            .remove(KEY_LAST_ERROR_TYPE)
+            .commit()
+    }
+
+    fun updateEndpoint(context: Context, scope: String, endpoint: PushEndpoint): Boolean {
+        val keys = endpoint.pubKeySet ?: return false
+        SubscriptionsDB(context).use { db ->
+            db.updateEndpoint(scope, endpoint.url, keys.pubKey, keys.auth)
+        }
+        clearError(context, scope)
+        return true
+    }
+
+    fun removeEndpoint(context: Context, scope: String) {
+        SubscriptionsDB(context).use { it.removeEndpoint(scope) }
+        clearError(context, scope)
+    }
+
+    fun errorType(error: PushError): String = when (error) {
+        is PushError.DB -> "database"
+        is PushError.Network -> "network"
+        is PushError.Registration -> "registration"
+        is PushError.ServiceUnavailable -> "service_unavailable"
+    }
+
+    fun prefs(context: Context): SharedPreferences =
+        context.getSharedPreferences(PREFS_NAME, Context.MODE_PRIVATE)
+}

packages/flutter_mozilla_components/android/src/main/kotlin/eu/weblibre/flutter_mozilla_components/push/UnifiedPushReceiver.kt

@@ -0,0 +1,173 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
+
+package eu.weblibre.flutter_mozilla_components.push
+
+import android.content.Context
+import android.content.Intent
+import android.util.Log
+import eu.weblibre.flutter_mozilla_components.ActiveProfile
+import eu.weblibre.flutter_mozilla_components.GlobalComponents
+import eu.weblibre.flutter_mozilla_components.ProfileContext
+import java.security.MessageDigest
+import java.util.UUID
+import java.util.concurrent.Executors
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.suspendCancellableCoroutine
+import org.ironfoxoss.unifiedpush.PushError
+import org.unifiedpush.android.connector.FailedReason
+import org.unifiedpush.android.connector.MessagingReceiver
+import org.unifiedpush.android.connector.data.PushEndpoint
+import org.unifiedpush.android.connector.data.PushMessage
+
+class UnifiedPushReceiver : MessagingReceiver() {
+    override fun onReceive(context: Context, intent: Intent) {
+        val action = intent.action
+        val token = runCatching { intent.getStringExtra(EXTRA_TOKEN) }.getOrNull()
+        if (action !in SUPPORTED_ACTIONS || token.isNullOrBlank()) {
+            Log.w(TAG, "Ignoring invalid UnifiedPush broadcast")
+            return
+        }
+
+        val pendingResult = goAsync()
+        synchronized(submissionLock) {
+            executor.execute {
+                try {
+                    val profileContext = ActiveProfile.resolveContext(context.applicationContext)
+                    if (profileContext == null) {
+                        Log.e(TAG, "UnifiedPush broadcast has no active profile")
+                        return@execute
+                    }
+                    currentToken.set(token)
+                    currentMessageId.set(runCatching { intent.getStringExtra(EXTRA_MESSAGE_ID) }.getOrNull())
+                    super.onReceive(profileContext, intent)
+                } catch (error: Throwable) {
+                    // An exception from onMessage deliberately prevents the connector from ACKing.
+                    Log.e(TAG, "UnifiedPush broadcast processing failed", error)
+                } finally {
+                    currentToken.remove()
+                    currentMessageId.remove()
+                    pendingResult.finish()
+                }
+            }
+        }
+    }
+
+    override fun onMessage(context: Context, message: PushMessage, instance: String) {
+        check(message.decrypted) { "Refusing to ACK an undecrypted push message" }
+        val profileContext = context as? ProfileContext
+            ?: error("UnifiedPush message did not use a profile context")
+        val id = durableMessageId(instance, checkNotNull(currentToken.get()), currentMessageId.get())
+        val stored = PushMessageStore(profileContext).persist(instance, message.content, id)
+        // MessagingReceiver sends its connector ACK only after this callback returns.
+        try {
+            PushMessageScheduler.enqueue(profileContext, stored.id)
+        } catch (error: Throwable) {
+            PushMessageScheduler.recoverLater(profileContext)
+            throw error
+        }
+    }
+
+    override fun onNewEndpoint(context: Context, endpoint: PushEndpoint, instance: String) {
+        val push = GlobalComponents.pushForProfile(context)
+        if (push != null) {
+            runBlocking { push.onNewEndpoint(instance, endpoint) }
+            push.emitStatusChanged()
+            return
+        }
+        if (PushProfileState.updateEndpoint(context, instance, endpoint)) {
+            Log.i(TAG, "Persisted endpoint for cold profile callback")
+        }
+    }
+
+    override fun onRegistrationFailed(context: Context, reason: FailedReason, instance: String) {
+        val error = reason.toPushError()
+        Log.w(TAG, "Push registration failed: ${error.message}")
+        val push = GlobalComponents.pushForProfile(context)
+        if (push != null) {
+            runBlocking { push.recordRegistrationError(instance, error) }
+            push.emitStatusChanged()
+        } else {
+            PushProfileState.recordError(
+                context,
+                instance,
+                PushProfileState.errorType(error),
+                error.message,
+            )
+        }
+    }
+
+    override fun onTempUnavailable(context: Context, instance: String) {
+        val push = GlobalComponents.pushForProfile(context)
+        if (push != null) {
+            runBlocking { push.recordTemporaryUnavailable(instance) }
+            push.emitStatusChanged()
+        } else {
+            PushProfileState.recordTemporaryUnavailable(context, instance)
+        }
+    }
+
+    override fun onUnregistered(context: Context, instance: String) {
+        val push = GlobalComponents.pushForProfile(context)
+        if (push != null) {
+            runBlocking { push.onUnregistered(instance) }
+            push.emitStatusChanged()
+        } else {
+            PushProfileState.removeEndpoint(context, instance)
+        }
+    }
+
+    private fun FailedReason.toPushError(): PushError = when (this) {
+        FailedReason.NETWORK -> PushError.Network("Push service needs network to register")
+        FailedReason.INTERNAL_ERROR -> PushError.ServiceUnavailable("Unknown error")
+        FailedReason.ACTION_REQUIRED ->
+            PushError.ServiceUnavailable("Push service waits for a user action")
+        FailedReason.VAPID_REQUIRED -> PushError.Registration("Push service requires VAPID")
+    }
+
+    companion object {
+        private const val TAG = "UnifiedPushReceiver"
+        private const val EXTRA_TOKEN = "token"
+        private const val EXTRA_MESSAGE_ID = "id"
+        private val SUPPORTED_ACTIONS = setOf(
+            "org.unifiedpush.android.connector.MESSAGE",
+            "org.unifiedpush.android.connector.UNREGISTERED",
+            "org.unifiedpush.android.connector.NEW_ENDPOINT",
+            "org.unifiedpush.android.connector.REGISTRATION_FAILED",
+            "org.unifiedpush.android.connector.TEMP_UNAVAILABLE",
+        )
+        private val executor = Executors.newSingleThreadExecutor()
+        private val submissionLock = Any()
+        private val currentToken = ThreadLocal<String?>()
+        private val currentMessageId = ThreadLocal<String?>()
+
+        internal fun durableMessageId(
+            scope: String,
+            connectorToken: String,
+            connectorId: String?,
+        ): String {
+            if (connectorId == null) return UUID.randomUUID().toString()
+            return MessageDigest.getInstance("SHA-256")
+                .digest("$scope\u0000$connectorToken\u0000$connectorId".toByteArray())
+                .joinToString("") { "%02x".format(it) }
+        }
+
+        internal suspend fun <T> runExclusive(block: suspend () -> T): T =
+            suspendCancellableCoroutine { continuation ->
+                synchronized(submissionLock) {
+                    val operationJob = Job(continuation.context[Job])
+                    val future = executor.submit {
+                        val result = runCatching { runBlocking(operationJob) { block() } }
+                        operationJob.complete()
+                        if (continuation.isActive) continuation.resumeWith(result)
+                    }
+                    continuation.invokeOnCancellation {
+                        operationJob.cancel()
+                        future.cancel(true)
+                    }
+                }
+            }
+    }
+}

packages/flutter_mozilla_components/android/src/main/kotlin/eu/weblibre/flutter_mozilla_components/push/WebNotificationDrainCoordinator.kt

@@ -0,0 +1,95 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
+
+package eu.weblibre.flutter_mozilla_components.push
+
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.Deferred
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.withTimeoutOrNull
+import mozilla.components.concept.engine.webnotifications.WebNotification
+import mozilla.components.concept.engine.webnotifications.WebNotificationDelegate
+import mozilla.components.support.ktx.kotlin.getOrigin
+
+/**
+ * Wraps the engine's real [WebNotificationDelegate] so a headless push delivery
+ * can stay alive until the service worker actually posts its notification.
+ *
+ * The web push handoff to Gecko returns no completion signal, so the service
+ * worker's `event.waitUntil(... showNotification())` runs entirely
+ * asynchronously. This coordinator lets the delivery observe the actual
+ * [onShowNotification] callback instead of guessing a duration, forwarding the
+ * notification to the real delegate unchanged.
+ */
+class WebNotificationDrainCoordinator : WebNotificationDelegate {
+    @Volatile
+    var delegate: WebNotificationDelegate? = null
+
+    private val lock = Any()
+    private var waiter: CompletableDeferred<Unit>? = null
+    private var waitOrigin: String? = null
+
+    override fun onShowNotification(webNotification: WebNotification): Deferred<Boolean> {
+        signal(webNotification.sourceUrl?.getOrigin())
+        // Preserve the engine's completion contract by returning the real
+        // delegate's deferred; only fall back if wrapping failed.
+        return delegate?.onShowNotification(webNotification) ?: CompletableDeferred(false)
+    }
+
+    override fun onCloseNotification(webNotification: WebNotification) {
+        delegate?.onCloseNotification(webNotification)
+    }
+
+    /**
+     * Run [deliver] (the push handoff to Gecko) and then keep the caller
+     * suspended until a matching web notification is shown or [timeoutMillis]
+     * elapses. When a notification is observed, wait a further [graceMillis] so
+     * the delegate's asynchronous `notify` can land before the caller returns
+     * and the process loses foreground priority.
+     *
+     * Origin matching is best-effort: if either the push [origin] or the
+     * notification's origin cannot be derived, any shown notification satisfies
+     * the wait. Deliveries are serialized under the profile lock, so at most one
+     * drain is armed at a time.
+     */
+    suspend fun drainWhileDelivering(
+        origin: String?,
+        timeoutMillis: Long,
+        graceMillis: Long,
+        deliver: suspend () -> Unit,
+    ) {
+        val deferred = CompletableDeferred<Unit>()
+        synchronized(lock) {
+            waiter = deferred
+            waitOrigin = origin
+        }
+        try {
+            deliver()
+            val shown = withTimeoutOrNull(timeoutMillis) {
+                deferred.await()
+                true
+            } == true
+            if (shown && graceMillis > 0) {
+                delay(graceMillis)
+            }
+        } finally {
+            synchronized(lock) {
+                if (waiter === deferred) {
+                    waiter = null
+                    waitOrigin = null
+                }
+            }
+        }
+    }
+
+    private fun signal(origin: String?) {
+        synchronized(lock) {
+            val pending = waiter ?: return
+            val target = waitOrigin
+            if (target == null || origin == null || target == origin) {
+                pending.complete(Unit)
+            }
+        }
+    }
+}

packages/flutter_mozilla_components/android/src/main/kotlin/eu/weblibre/flutter_mozilla_components/push/WebPushEngineIntegration.kt

@@ -6,8 +6,11 @@ package eu.weblibre.flutter_mozilla_components.push
 
 import android.util.Base64
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.MainScope
+import kotlinx.coroutines.cancel
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.withContext
 import mozilla.components.concept.engine.Engine
 import mozilla.components.concept.engine.webpush.WebPushDelegate
 import mozilla.components.concept.engine.webpush.WebPushHandler
@@ -44,6 +47,25 @@ class WebPushEngineIntegration(
         pushFeature.unregister(this)
     }
 
+    suspend fun deliverMessage(scope: PushScope, payload: ByteArray?) {
+        withContext(Dispatchers.Main.immediate) {
+            checkNotNull(handler) { "Web push handler is not initialized" }
+                .onPushMessage(scope, payload)
+        }
+    }
+
+    suspend fun invalidateEndpoint(scope: PushScope) {
+        withContext(Dispatchers.Main.immediate) {
+            handler?.onSubscriptionChanged(scope)
+        }
+    }
+
+    fun close() {
+        stop()
+        handler = null
+        coroutineScope.cancel()
+    }
+
     override fun onMessageReceived(scope: PushScope, message: ByteArray?) {
         coroutineScope.launch {
             handler?.onPushMessage(scope, message)

packages/flutter_mozilla_components/android/src/main/kotlin/eu/weblibre/flutter_mozilla_components/receivers/UnifiedPushReceiver.kt

@@ -1,75 +0,0 @@
-/* This Source Code Form is subject to the terms of the Mozilla Public
- * License, v. 2.0. If a copy of the MPL was not distributed with this
- * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
-
-package eu.weblibre.flutter_mozilla_components.receivers
-
-import android.content.Context
-import android.content.Intent
-import android.util.Log
-import eu.weblibre.flutter_mozilla_components.ActiveProfile
-import eu.weblibre.flutter_mozilla_components.GlobalComponents
-import org.ironfoxoss.unifiedpush.PushError
-import org.ironfoxoss.unifiedpush.UnifiedPushProcessor
-import org.unifiedpush.android.connector.FailedReason
-import org.unifiedpush.android.connector.MessagingReceiver
-import org.unifiedpush.android.connector.data.PushEndpoint
-import org.unifiedpush.android.connector.data.PushMessage
-
-class UnifiedPushReceiver : MessagingReceiver() {
-    companion object {
-        private const val TAG = "UnifiedPushReceiver"
-    }
-
-    override fun onReceive(context: Context, intent: Intent) {
-        ActiveProfile.resolveFromDisk(context.applicationContext)
-
-        if (GlobalComponents.components == null &&
-            !GlobalComponents.ensureExternalComponents(context.applicationContext)
-        ) {
-            Log.e(TAG, "Unable to initialize components for UnifiedPush delivery")
-            return
-        }
-
-        GlobalComponents.components?.push?.initialize()
-
-        if (GlobalComponents.components == null) {
-            Log.e(TAG, "UnifiedPush delivery aborted because components are unavailable")
-            return
-        }
-
-        super.onReceive(context, intent)
-    }
-
-    override fun onMessage(context: Context, message: PushMessage, instance: String) {
-        UnifiedPushProcessor.requireInstance.onMessage(
-            scope = instance,
-            message = message,
-        )
-    }
-
-    override fun onNewEndpoint(context: Context, endpoint: PushEndpoint, instance: String) {
-        UnifiedPushProcessor.requireInstance.onNewEndpoint(
-            scope = instance,
-            newEndpoint = endpoint,
-        )
-    }
-
-    override fun onRegistrationFailed(context: Context, reason: FailedReason, instance: String) {
-        UnifiedPushProcessor.requireInstance.onError(reason.toPushError())
-    }
-
-    override fun onUnregistered(context: Context, instance: String) {
-        UnifiedPushProcessor.requireInstance.onUnregistered(scope = instance)
-    }
-
-    private fun FailedReason.toPushError(): PushError {
-        return when (this) {
-            FailedReason.NETWORK -> PushError.Network("Push service needs network to register")
-            FailedReason.INTERNAL_ERROR -> PushError.ServiceUnavailable("Unknown error")
-            FailedReason.ACTION_REQUIRED ->
-                PushError.ServiceUnavailable("Push service waits for a user action")
-            FailedReason.VAPID_REQUIRED -> PushError.Registration("Push service requires VAPID")
-        }
-    }
-}

packages/flutter_mozilla_components/android/src/test/kotlin/eu/lensai/flutter_mozilla_components/FlutterMozillaContextPluginTest.kt

@@ -1,27 +1,11 @@
 package eu.weblibre.flutter_mozilla_components
 
-import io.flutter.plugin.common.MethodCall
-import io.flutter.plugin.common.MethodChannel
 import kotlin.test.Test
-import org.mockito.Mockito
-
-/*
- * This demonstrates a simple unit test of the Kotlin portion of this plugin's implementation.
- *
- * Once you have built the plugin's example app, you can run these tests from the command
- * line by running `./gradlew testDebugUnitTest` in the `example/android/` directory, or
- * you can run them directly from IDEs that support JUnit such as Android Studio.
- */
+import kotlin.test.assertNotNull
 
 internal class FlutterMozillaContextPluginTest {
-  @Test
-  fun onMethodCall_getPlatformVersion_returnsExpectedValue() {
-    val plugin = FlutterMozillaComponentsPlugin()
-
-    val call = MethodCall("getPlatformVersion", null)
-    val mockResult: MethodChannel.Result = Mockito.mock(MethodChannel.Result::class.java)
-    plugin.onMethodCall(call, mockResult)
-
-    Mockito.verify(mockResult).success("Android " + android.os.Build.VERSION.RELEASE)
-  }
+    @Test
+    fun pluginCanBeConstructed() {
+        assertNotNull(FlutterMozillaComponentsPlugin())
+    }
 }

packages/flutter_mozilla_components/android/src/test/kotlin/eu/weblibre/flutter_mozilla_components/ActiveProfileTest.kt

@@ -0,0 +1,56 @@
+/*
+ * Copyright (c) 2024-2025 Fabian Freund.
+ *
+ * This file is part of WebLibre
+ * (see https://weblibre.eu).
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+package eu.weblibre.flutter_mozilla_components
+
+import kotlin.test.Test
+import kotlin.test.assertFalse
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import kotlinx.coroutines.withTimeoutOrNull
+
+class ActiveProfileTest {
+    @Test
+    fun profileLockIsRetainedAcrossSuspension() = runBlocking {
+        val entered = CompletableDeferred<Unit>()
+        val release = CompletableDeferred<Unit>()
+        val secondEntered = CompletableDeferred<Unit>()
+        val first = launch {
+            ActiveProfile.withProfileLock {
+                entered.complete(Unit)
+                release.await()
+            }
+        }
+
+        entered.await()
+        val second = launch {
+            ActiveProfile.withProfileLock { secondEntered.complete(Unit) }
+        }
+
+        withTimeoutOrNull(100) { secondEntered.await() }
+        assertFalse(secondEntered.isCompleted)
+        release.complete(Unit)
+        withTimeout(1_000) {
+            first.join()
+            second.join()
+        }
+    }
+}

packages/flutter_mozilla_components/android/src/test/kotlin/eu/weblibre/flutter_mozilla_components/push/ProfileSwitchTimeoutTest.kt

@@ -0,0 +1,52 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
+
+package eu.weblibre.flutter_mozilla_components.push
+
+import kotlin.test.Test
+import kotlin.test.assertFalse
+import kotlin.test.assertTrue
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.async
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+
+class ProfileSwitchTimeoutTest {
+    @Test
+    fun timeoutPreventsOperationFromStarting() = runBlocking {
+        var sideEffectRan = false
+
+        val completed = runWithStartTimeout(50) { tryStart ->
+            delay(200)
+            if (tryStart()) sideEffectRan = true
+        }
+
+        assertFalse(completed)
+        assertFalse(sideEffectRan)
+    }
+
+    @Test
+    fun operationCompletesAfterStartingBeforeTimeout() = runBlocking {
+        val started = CompletableDeferred<Unit>()
+        val release = CompletableDeferred<Unit>()
+        var sideEffectRan = false
+        val result = async {
+            runWithStartTimeout(50) { tryStart ->
+                assertTrue(tryStart())
+                started.complete(Unit)
+                release.await()
+                sideEffectRan = true
+            }
+        }
+
+        started.await()
+        delay(100)
+        assertFalse(result.isCompleted)
+
+        release.complete(Unit)
+        assertTrue(withTimeout(1_000) { result.await() })
+        assertTrue(sideEffectRan)
+    }
+}

packages/flutter_mozilla_components/android/src/test/kotlin/eu/weblibre/flutter_mozilla_components/push/PushMessageStoreTest.kt

@@ -0,0 +1,132 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
+
+package eu.weblibre.flutter_mozilla_components.push
+
+import java.nio.file.Files
+import kotlin.io.path.createTempDirectory
+import kotlin.test.Test
+import kotlin.test.assertContentEquals
+import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
+import kotlin.test.assertFalse
+import kotlin.test.assertNull
+import kotlin.test.assertTrue
+
+class PushMessageStoreTest {
+    @Test
+    fun persistsListsAndDeletesMessage() {
+        val directory = createTempDirectory("push-store").toFile()
+        try {
+            val store = PushMessageStore(directory)
+            store.persist("https://example.com", byteArrayOf(0, 1, 2, -1), "message-1")
+
+            assertEquals(listOf("message-1"), store.ids())
+            val stored = store.get("message-1")
+            assertEquals("https://example.com", stored?.scope)
+            assertContentEquals(byteArrayOf(0, 1, 2, -1), stored?.payload)
+            assertTrue(store.complete("message-1"))
+            assertNull(store.get("message-1"))
+        } finally {
+            directory.deleteRecursively()
+        }
+    }
+
+    @Test
+    fun replacingIdLeavesOneCompleteRecord() {
+        val directory = createTempDirectory("push-store").toFile()
+        try {
+            val store = PushMessageStore(directory)
+            store.persist("old", byteArrayOf(1), "same-id")
+            store.persist("new", byteArrayOf(2, 3), "same-id")
+
+            assertEquals(listOf("same-id"), store.ids())
+            assertEquals("new", store.get("same-id")?.scope)
+            assertContentEquals(byteArrayOf(2, 3), store.get("same-id")?.payload)
+        } finally {
+            directory.deleteRecursively()
+        }
+    }
+
+    @Test
+    fun rejectsPathTraversalIds() {
+        val directory = Files.createTempDirectory("push-store").toFile()
+        try {
+            val store = PushMessageStore(directory)
+            assertFailsWith<IllegalArgumentException> {
+                store.persist("scope", byteArrayOf(1), "../outside")
+            }
+        } finally {
+            directory.deleteRecursively()
+        }
+    }
+
+    @Test
+    fun completedMessageIsNotRecovered() {
+        val directory = createTempDirectory("push-store").toFile()
+        try {
+            val store = PushMessageStore(directory)
+            store.persist("scope", byteArrayOf(1), "completed")
+
+            assertTrue(store.complete("completed"))
+
+            assertTrue(store.ids().isEmpty())
+            assertNull(store.get("completed"))
+            assertFalse(directory.resolve("completed.push").exists())
+            assertTrue(directory.resolve("completed.delivered").isFile)
+        } finally {
+            directory.deleteRecursively()
+        }
+    }
+
+    @Test
+    fun rejectsAndDiscardsCorruptMessage() {
+        val directory = createTempDirectory("push-store").toFile()
+        try {
+            val store = PushMessageStore(directory)
+            directory.resolve("corrupt.push").writeBytes(byteArrayOf(1, 2, 3))
+
+            assertFailsWith<CorruptPushMessageException> {
+                store.get("corrupt")
+            }
+            assertTrue(store.complete("corrupt"))
+            assertTrue(store.ids().isEmpty())
+        } finally {
+            directory.deleteRecursively()
+        }
+    }
+
+    @Test
+    fun deliveredMarkerSuppressesStalePayload() {
+        val directory = createTempDirectory("push-store").toFile()
+        try {
+            val store = PushMessageStore(directory)
+            store.persist("scope", byteArrayOf(1), "stale")
+            directory.resolve("stale.delivered").writeBytes(byteArrayOf(1))
+
+            assertTrue(store.ids().isEmpty())
+            assertNull(store.get("stale"))
+        } finally {
+            directory.deleteRecursively()
+        }
+    }
+
+    @Test
+    fun expiredDeliveredMarkerAllowsMessageIdReuse() {
+        val directory = createTempDirectory("push-store").toFile()
+        try {
+            val store = PushMessageStore(directory)
+            store.persist("old", byteArrayOf(1), "reused")
+            assertTrue(store.complete("reused"))
+            assertTrue(directory.resolve("reused.delivered").setLastModified(0))
+
+            store.persist("new", byteArrayOf(2), "reused")
+
+            assertEquals(listOf("reused"), store.ids())
+            assertEquals("new", store.get("reused")?.scope)
+        } finally {
+            directory.deleteRecursively()
+        }
+    }
+}

packages/flutter_mozilla_components/android/src/test/kotlin/eu/weblibre/flutter_mozilla_components/push/UnifiedPushReceiverTest.kt

@@ -0,0 +1,79 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
+
+package eu.weblibre.flutter_mozilla_components.push
+
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+import kotlin.test.assertNotEquals
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.awaitCancellation
+import kotlinx.coroutines.cancelAndJoin
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import kotlinx.coroutines.withTimeoutOrNull
+
+class UnifiedPushReceiverTest {
+    @Test
+    fun durableIdIsStableWithinConnectorRegistration() {
+        val first = UnifiedPushReceiver.durableMessageId("scope", "token", "message")
+        val second = UnifiedPushReceiver.durableMessageId("scope", "token", "message")
+
+        assertEquals(first, second)
+    }
+
+    @Test
+    fun durableIdSeparatesConnectorRegistrations() {
+        val first = UnifiedPushReceiver.durableMessageId("scope", "old-token", "message")
+        val second = UnifiedPushReceiver.durableMessageId("scope", "new-token", "message")
+
+        assertNotEquals(first, second)
+    }
+
+    @Test
+    fun cancellingExclusiveOperationReleasesQueue() = runBlocking {
+        val entered = CompletableDeferred<Unit>()
+        val operation = launch {
+            UnifiedPushReceiver.runExclusive {
+                entered.complete(Unit)
+                awaitCancellation()
+            }
+        }
+
+        entered.await()
+        operation.cancelAndJoin()
+
+        withTimeout(1_000) {
+            UnifiedPushReceiver.runExclusive { }
+        }
+    }
+
+    @Test
+    fun exclusiveOperationRetainsQueueAcrossSuspension() = runBlocking {
+        val entered = CompletableDeferred<Unit>()
+        val release = CompletableDeferred<Unit>()
+        val secondEntered = CompletableDeferred<Unit>()
+        val first = launch {
+            UnifiedPushReceiver.runExclusive {
+                entered.complete(Unit)
+                release.await()
+            }
+        }
+
+        entered.await()
+        val second = launch {
+            UnifiedPushReceiver.runExclusive { secondEntered.complete(Unit) }
+        }
+
+        withTimeoutOrNull(100) { secondEntered.await() }
+        assertFalse(secondEntered.isCompleted)
+        release.complete(Unit)
+        withTimeout(1_000) {
+            first.join()
+            second.join()
+        }
+    }
+}

packages/flutter_mozilla_components/lib/flutter_mozilla_components.dart

@@ -26,6 +26,7 @@ export 'src/domain/services/gecko_icon.dart';
 export 'src/domain/services/gecko_logging.dart';
 export 'src/domain/services/gecko_ml.dart';
 export 'src/domain/services/gecko_pref.dart';
+export 'src/domain/services/gecko_push.dart';
 export 'src/domain/services/gecko_readerable.dart';
 export 'src/domain/services/gecko_selection_action.dart';
 export 'src/domain/services/gecko_session.dart';
@@ -100,6 +101,10 @@ export 'src/pigeons/gecko.g.dart'
         MlProgressType,
         PhoneHitResult,
         ProxyLoadError,
+        PushDistributor,
+        PushDistributorStatus,
+        PushStatus,
+        PushSubscription,
         PwaIcon,
         PwaManifest,
         QueryParameterStripping,

packages/flutter_mozilla_components/lib/src/domain/services/gecko_browser.dart

@@ -69,10 +69,6 @@ class GeckoBrowserService {
     return _api.requestDefaultBrowser();
   }
 
-  Future<bool> pickUnifiedPushDistributor() {
-    return _api.pickUnifiedPushDistributor();
-  }
-
   Future<void> shutdown() {
     return _api.shutdown();
   }

packages/flutter_mozilla_components/lib/src/domain/services/gecko_push.dart

@@ -0,0 +1,115 @@
+/*
+ * This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/.
+ */
+
+import 'package:flutter/services.dart';
+import 'package:flutter_mozilla_components/src/pigeons/gecko.g.dart';
+import 'package:rxdart/rxdart.dart';
+
+/// Service for UnifiedPush-backed web push.
+///
+/// Web push is delivered by a separate distributor app (ntfy, Sunup, …) that the
+/// user selects. With no distributor selected nothing can be delivered, so the
+/// distributor selection doubles as the on/off switch for web push.
+///
+/// Subscriptions are exposed read-only: Gecko owns the subscription state and
+/// offers no app-facing channel to revoke one, so removal must go through the
+/// site's notification permission.
+class GeckoPushService extends GeckoPushEvents {
+  final GeckoPushApi _api;
+  final BinaryMessenger? _defaultBinaryMessenger;
+  final String _defaultMessageChannelSuffix;
+
+  final _statusSubject = PublishSubject<PushStatus>();
+  BinaryMessenger? _eventBinaryMessenger;
+  String _eventMessageChannelSuffix = '';
+  int? _lastStatusSequence;
+  bool _isSetUp = false;
+  bool _disposed = false;
+  Future<void>? _disposeFuture;
+
+  /// Stream of status snapshots pushed from native, emitted when a distributor
+  /// acknowledges registration, fails to register, or is uninstalled.
+  ///
+  /// Non-replaying: callers that need the current value must subscribe to this
+  /// before calling [getPushStatus], or they will miss any transition that lands
+  /// between the two.
+  Stream<PushStatus> get statusChanges => _statusSubject.stream;
+
+  GeckoPushService({
+    BinaryMessenger? binaryMessenger,
+    String messageChannelSuffix = '',
+  }) : _defaultBinaryMessenger = binaryMessenger,
+       _defaultMessageChannelSuffix = messageChannelSuffix,
+       _api = GeckoPushApi(
+         binaryMessenger: binaryMessenger,
+         messageChannelSuffix: messageChannelSuffix,
+       );
+
+  /// Sets up the service to receive events from native.
+  ///
+  /// Must be called before events will be received.
+  void setUp({BinaryMessenger? binaryMessenger, String? messageChannelSuffix}) {
+    if (_isSetUp || _disposed) {
+      return;
+    }
+
+    _eventBinaryMessenger = binaryMessenger ?? _defaultBinaryMessenger;
+    _eventMessageChannelSuffix =
+        messageChannelSuffix ?? _defaultMessageChannelSuffix;
+    GeckoPushEvents.setUp(
+      this,
+      binaryMessenger: _eventBinaryMessenger,
+      messageChannelSuffix: _eventMessageChannelSuffix,
+    );
+    _isSetUp = true;
+  }
+
+  Future<PushStatus> getPushStatus() => _api.getPushStatus();
+
+  /// Selects [packageName], which must be one of [PushStatus.available].
+  Future<void> setDistributor(String packageName) =>
+      _api.setDistributor(packageName);
+
+  /// Forgets the current distributor, disabling web push delivery.
+  Future<void> removeDistributor() => _api.removeDistributor();
+
+  Future<void> renewRegistration() => _api.renewRegistration();
+
+  Future<void> suspendForProfileSwitch(String targetProfileId) =>
+      _api.suspendForProfileSwitch(targetProfileId);
+
+  Future<List<PushSubscription>> getSubscriptions() => _api.getSubscriptions();
+
+  // GeckoPushEvents implementation
+
+  @override
+  void onPushStatusChanged(int sequence, PushStatus status) {
+    if (_disposed ||
+        (_lastStatusSequence != null && sequence <= _lastStatusSequence!)) {
+      return;
+    }
+
+    _lastStatusSequence = sequence;
+    _statusSubject.add(status);
+  }
+
+  Future<void> dispose() {
+    return _disposeFuture ??= _dispose();
+  }
+
+  Future<void> _dispose() async {
+    _disposed = true;
+    if (_isSetUp) {
+      GeckoPushEvents.setUp(
+        null,
+        binaryMessenger: _eventBinaryMessenger,
+        messageChannelSuffix: _eventMessageChannelSuffix,
+      );
+      _isSetUp = false;
+    }
+    await _statusSubject.close();
+  }
+}

packages/flutter_mozilla_components/pigeons/gecko.dart

@@ -1417,8 +1417,6 @@ abstract class GeckoBrowserApi {
   });
   bool isDefaultBrowser();
   void requestDefaultBrowser();
-  @async
-  bool pickUnifiedPushDistributor();
   void shutdown();
 }
 
@@ -3146,3 +3144,105 @@ abstract class GeckoGestureEvents {
   /// [sequence] Event sequence number for ordering.
   void onGestureReset(int sequence);
 }
+
+/// Lifecycle state of the selected UnifiedPush distributor.
+enum PushDistributorStatus {
+  /// No distributor app is installed on the device.
+  noneAvailable,
+
+  /// Distributors are installed but the user has not chosen one.
+  notSelected,
+
+  /// A distributor is chosen but has not acknowledged our registration yet.
+  pending,
+
+  /// A distributor is chosen and has acknowledged our registration.
+  ready,
+
+  /// A distributor was chosen previously but is no longer installed. Web push
+  /// is dead in this state and there is no fallback transport.
+  unavailable,
+}
+
+class PushDistributor {
+  final String packageName;
+
+  /// Human-readable app label, or null if the package is no longer installed.
+  final String? label;
+
+  PushDistributor({required this.packageName, required this.label});
+}
+
+class PushStatus {
+  final PushDistributorStatus status;
+  final PushDistributor? current;
+  final List<PushDistributor> available;
+
+  /// Most recent distributor registration failure, or null if none.
+  ///
+  /// Held natively rather than delivered as a one-shot event: registrations are
+  /// attempted at startup and from background broadcasts, both of which can run
+  /// long before any Dart listener exists.
+  final String? lastError;
+
+  PushStatus({
+    required this.status,
+    required this.current,
+    required this.available,
+    required this.lastError,
+  });
+}
+
+class PushSubscription {
+  /// Subscription identifier, which for web push is the site's origin.
+  final String scope;
+
+  /// Whether the distributor has handed back an endpoint for this scope.
+  final bool hasEndpoint;
+
+  PushSubscription({required this.scope, required this.hasEndpoint});
+}
+
+/// Dart → Kotlin. UnifiedPush distributor management and web push introspection.
+@HostApi()
+abstract class GeckoPushApi {
+  @async
+  PushStatus getPushStatus();
+
+  /// Selects [packageName], which must be one of [PushStatus.available].
+  ///
+  /// The picker is built in Dart rather than delegated to the connector's own
+  /// dialog, which would save the selection against a non-profile context.
+  @async
+  void setDistributor(String packageName);
+
+  /// Forgets the current distributor. This is the off switch for web push.
+  @async
+  void removeDistributor();
+
+  @async
+  void renewRegistration();
+
+  /// Pauses push transport for the current profile before switching profiles.
+  /// Site subscriptions and the chosen distributor are retained for restoration
+  /// when this profile becomes active again.
+  @async
+  void suspendForProfileSwitch(String targetProfileId);
+
+  /// Subscriptions Gecko has created, read from the UnifiedPush store. Read-only:
+  /// there is no app→Gecko channel to revoke a subscription, so removal has to go
+  /// through the site's notification permission instead.
+  @async
+  List<PushSubscription> getSubscriptions();
+}
+
+/// Kotlin → Dart. Push registration lifecycle.
+///
+/// Registration failures reach Dart through [PushStatus.lastError] rather than a
+/// dedicated event, so a failure raised before any Dart listener is attached is
+/// still visible the first time the settings screen reads the status.
+@FlutterApi()
+abstract class GeckoPushEvents {
+  /// [sequence] Event sequence number for ordering.
+  void onPushStatusChanged(int sequence, PushStatus status);
+}

packages/flutter_mozilla_components/test/gecko_push_test.dart

@@ -0,0 +1,96 @@
+import 'dart:async';
+import 'dart:typed_data';
+
+import 'package:flutter/services.dart';
+import 'package:flutter_mozilla_components/flutter_mozilla_components.dart';
+import 'package:flutter_mozilla_components/src/pigeons/gecko.g.dart'
+    show GeckoPushEvents;
+import 'package:flutter_test/flutter_test.dart';
+
+void main() {
+  TestWidgetsFlutterBinding.ensureInitialized();
+
+  test('accepts sequence zero and ignores duplicate or older events', () async {
+    final service = GeckoPushService();
+    addTearDown(service.dispose);
+    final statuses = <PushStatus>[];
+    final subscription = service.statusChanges.listen(statuses.add);
+    addTearDown(subscription.cancel);
+
+    service.onPushStatusChanged(0, _status(PushDistributorStatus.pending));
+    service.onPushStatusChanged(0, _status(PushDistributorStatus.ready));
+    service.onPushStatusChanged(-1, _status(PushDistributorStatus.unavailable));
+    service.onPushStatusChanged(2, _status(PushDistributorStatus.ready));
+    await pumpEventQueue();
+
+    expect(statuses.map((status) => status.status), [
+      PushDistributorStatus.pending,
+      PushDistributorStatus.ready,
+    ]);
+  });
+
+  test(
+    'setup and disposal are idempotent and unregister the exact channel',
+    () async {
+      final messenger =
+          TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger;
+      final service = GeckoPushService(
+        binaryMessenger: messenger,
+        messageChannelSuffix: 'push-test',
+      );
+      final statuses = <PushStatus>[];
+      final subscription = service.statusChanges.listen(statuses.add);
+      addTearDown(subscription.cancel);
+
+      service.setUp();
+      service.setUp();
+
+      final responseBeforeDispose = await _dispatchStatus(
+        messenger,
+        suffix: 'push-test',
+        sequence: 1,
+        status: _status(PushDistributorStatus.ready),
+      );
+      await pumpEventQueue();
+
+      expect(responseBeforeDispose, isNotNull);
+      expect(statuses, hasLength(1));
+
+      await service.dispose();
+      await service.dispose();
+
+      final responseAfterDispose = await _dispatchStatus(
+        messenger,
+        suffix: 'push-test',
+        sequence: 2,
+        status: _status(PushDistributorStatus.unavailable),
+      );
+      service.onPushStatusChanged(3, _status(PushDistributorStatus.pending));
+      service.setUp();
+      await pumpEventQueue();
+
+      expect(responseAfterDispose, isNull);
+      expect(statuses, hasLength(1));
+    },
+  );
+}
+
+PushStatus _status(PushDistributorStatus status) {
+  return PushStatus(status: status, available: const []);
+}
+
+Future<ByteData?> _dispatchStatus(
+  TestDefaultBinaryMessenger messenger, {
+  required String suffix,
+  required int sequence,
+  required PushStatus status,
+}) async {
+  final reply = Completer<ByteData?>();
+  final channelSuffix = suffix.isEmpty ? '' : '.$suffix';
+  await messenger.handlePlatformMessage(
+    'dev.flutter.pigeon.flutter_mozilla_components.GeckoPushEvents.onPushStatusChanged$channelSuffix',
+    GeckoPushEvents.pigeonChannelCodec.encodeMessage([sequence, status]),
+    reply.complete,
+  );
+  return reply.future;
+}