diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/OneSignalDispatchers.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/OneSignalDispatchers.kt
index 3b067820b1..56d8357a9d 100644
--- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/OneSignalDispatchers.kt
+++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/OneSignalDispatchers.kt
@@ -9,6 +9,8 @@ import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.asCoroutineDispatcher
 import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
+import java.util.concurrent.ExecutorService
+import java.util.concurrent.Executors
 import java.util.concurrent.LinkedBlockingQueue
 import java.util.concurrent.ThreadFactory
 import java.util.concurrent.ThreadPoolExecutor
@@ -43,6 +45,8 @@ object OneSignalDispatchers {
         "$BASE_THREAD_NAME-IO" // Thread name prefix for I/O operations
     private const val DEFAULT_THREAD_NAME_PREFIX =
         "$BASE_THREAD_NAME-Default" // Thread name prefix for CPU operations
+    private const val SERIAL_IO_THREAD_NAME =
+        "$BASE_THREAD_NAME-SerialIO" // Single, named thread for order-sensitive work
 
     private class OptimizedThreadFactory(
         private val namePrefix: String,
@@ -80,6 +84,27 @@ object OneSignalDispatchers {
         }
     }
 
+    /**
+     * Single-thread executor for order-sensitive work (e.g. lifecycle event handlers that
+     * must run in the order the events fired on the main thread). Submission order on the
+     * main thread equals execution order on this thread, so callers don't need their own
+     * synchronization to preserve sequence — they only need to capture any time-sensitive
+     * state (timestamps, current state snapshots) on the caller before dispatching.
+     */
+    private val serialIOExecutor: ExecutorService by lazy {
+        try {
+            Executors.newSingleThreadExecutor(
+                OptimizedThreadFactory(
+                    namePrefix = SERIAL_IO_THREAD_NAME,
+                    priority = Thread.NORM_PRIORITY - 1,
+                ),
+            )
+        } catch (e: Exception) {
+            Logging.error("OneSignalDispatchers: Failed to create SerialIO executor: ${e.message}")
+            throw e
+        }
+    }
+
     private val defaultExecutor: ThreadPoolExecutor by lazy {
         try {
             ThreadPoolExecutor(
@@ -117,6 +142,18 @@ object OneSignalDispatchers {
         }
     }
 
+    val SerialIO: CoroutineDispatcher by lazy {
+        try {
+            serialIOExecutor.asCoroutineDispatcher()
+        } catch (e: Exception) {
+            // Fall back to a limited-parallelism view of Dispatchers.IO. A single shared view
+            // serializes our submissions even when our own executor failed to start.
+            Logging.error("OneSignalDispatchers: Using fallback serialized Dispatchers.IO: ${e.message}")
+            @Suppress("OPT_IN_USAGE")
+            Dispatchers.IO.limitedParallelism(1)
+        }
+    }
+
     private val IOScope: CoroutineScope by lazy {
         CoroutineScope(SupervisorJob() + IO)
     }
@@ -125,6 +162,10 @@ object OneSignalDispatchers {
         CoroutineScope(SupervisorJob() + Default)
     }
 
+    private val SerialIOScope: CoroutineScope by lazy {
+        CoroutineScope(SupervisorJob() + SerialIO)
+    }
+
     fun launchOnIO(block: suspend () -> Unit): Job {
         return IOScope.launch { block() }
     }
@@ -133,16 +174,94 @@ object OneSignalDispatchers {
         return DefaultScope.launch { block() }
     }
 
+    /**
+     * Launches [block] on the dedicated serial IO thread. Tasks submitted from any thread
+     * run one-at-a-time in submission order, which makes this the right entry point for
+     * order-sensitive lifecycle work (e.g. focus/unfocus handlers).
+     */
+    fun launchOnSerialIO(block: suspend () -> Unit): Job {
+        return SerialIOScope.launch { block() }
+    }
+
+    @Volatile
+    private var prewarmStarted = false
+    private val prewarmLock = Any()
+
+    /**
+     * Triggers the lazy initialization of [IO], [Default], and [SerialIO] (and their backing
+     * executors, dispatchers, and scopes) on a short-lived background thread.
+     *
+     * Background:
+     * The lazy `by lazy` properties below construct `ThreadPoolExecutor` instances and wrap them
+     * in `asCoroutineDispatcher() + SupervisorJob() + CoroutineScope(...)`. Production OTel
+     * shows that when the **first** caller of [launchOnIO] / [launchOnSerialIO] is on the main
+     * thread (Activity-lifecycle handler, `JobService.onStartJob`, etc.), the construction cost
+     * — which includes a `kotlinx.coroutines.BuildersKt.launch` that hits
+     * `ThreadPoolExecutor.execute` and `LinkedBlockingQueue.offer` synchronously — is paid on
+     * the calling thread, blocking the main thread for many seconds on cold start. See SDK-4507.
+     *
+     * Calling [prewarm] from a non-time-sensitive spot in [com.onesignal.OneSignal.initWithContext]
+     * shifts that cost to a dedicated `OneSignal-prewarm` daemon thread, so the first
+     * production caller — including main-thread lifecycle handlers — only pays the much cheaper
+     * "submit work to an already-constructed executor" cost.
+     *
+     * Idempotent and fire-and-forget: a no-op on second and subsequent calls. Safe to invoke
+     * from any thread; the heavy lifting always happens on the daemon thread we spawn here, not
+     * on the caller. Failures are logged and swallowed because the executors retain their
+     * existing fallback paths (e.g. `Dispatchers.IO.limitedParallelism(1)` for [SerialIO]) and a
+     * failed prewarm will simply mean the first production caller pays the original cost.
+     */
+    fun prewarm() {
+        if (prewarmStarted) return
+        synchronized(prewarmLock) {
+            if (prewarmStarted) return
+            prewarmStarted = true
+        }
+        val prewarmThread = Thread(
+            {
+                try {
+                    // Each launch* call below triggers the corresponding lazy chain
+                    // (executor -> dispatcher -> scope) and submits an empty coroutine,
+                    // which forces the worker thread(s) to start as well.
+                    launchOnIO { /* warm IOScope + ioExecutor */ }
+                    launchOnDefault { /* warm DefaultScope + defaultExecutor */ }
+                    launchOnSerialIO { /* warm SerialIOScope + serialIOExecutor */ }
+                } catch (e: Exception) {
+                    Logging.warn("OneSignalDispatchers.prewarm failed: ${e.message}")
+                }
+            },
+            "$BASE_THREAD_NAME-prewarm",
+        )
+        prewarmThread.isDaemon = true
+        prewarmThread.priority = Thread.NORM_PRIORITY - 2
+        prewarmThread.start()
+    }
+
+    /**
+     * Test-only hook to reset [prewarmStarted] so different specs can exercise the
+     * "first call wins" branch independently. Not part of any public contract.
+     */
+    internal fun resetPrewarmForTest() {
+        synchronized(prewarmLock) {
+            prewarmStarted = false
+        }
+    }
+
     internal fun getPerformanceMetrics(): String {
         return try {
+            val serialQueueSize =
+                (serialIOExecutor as? ThreadPoolExecutor)?.queue?.size?.toString() ?: "n/a"
+            val serialCompleted =
+                (serialIOExecutor as? ThreadPoolExecutor)?.completedTaskCount ?: 0L
             """
             OneSignalDispatchers Performance Metrics:
             - IO Pool: ${ioExecutor.activeCount}/${ioExecutor.corePoolSize} active/core threads
             - IO Queue: ${ioExecutor.queue.size} pending tasks
             - Default Pool: ${defaultExecutor.activeCount}/${defaultExecutor.corePoolSize} active/core threads
             - Default Queue: ${defaultExecutor.queue.size} pending tasks
-            - Total completed tasks: ${ioExecutor.completedTaskCount + defaultExecutor.completedTaskCount}
-            - Memory usage: ~${(ioExecutor.activeCount + defaultExecutor.activeCount) * 1024}KB (thread stacks, ~1MB each)
+            - SerialIO Queue: $serialQueueSize pending tasks
+            - Total completed tasks: ${ioExecutor.completedTaskCount + defaultExecutor.completedTaskCount + serialCompleted}
+            - Memory usage: ~${(ioExecutor.activeCount + defaultExecutor.activeCount + 1) * 1024}KB (thread stacks, ~1MB each)
             """.trimIndent()
         } catch (e: Exception) {
             "OneSignalDispatchers not initialized or using fallback dispatchers ${e.message}"
@@ -150,40 +269,41 @@ object OneSignalDispatchers {
     }
 
     internal fun getStatus(): String {
-        val ioExecutorStatus =
-            try {
-                if (ioExecutor.isShutdown) "Shutdown" else "Active"
-            } catch (e: Exception) {
-                "ioExecutor Not initialized ${e.message ?: "Unknown error"}"
-            }
-
-        val defaultExecutorStatus =
-            try {
-                if (defaultExecutor.isShutdown) "Shutdown" else "Active"
-            } catch (e: Exception) {
-                "defaultExecutor Not initialized ${e.message ?: "Unknown error"}"
-            }
-
-        val ioScopeStatus =
-            try {
-                if (IOScope.isActive) "Active" else "Cancelled"
-            } catch (e: Exception) {
-                "IOScope Not initialized ${e.message ?: "Unknown error"}"
-            }
-
-        val defaultScopeStatus =
-            try {
-                if (DefaultScope.isActive) "Active" else "Cancelled"
-            } catch (e: Exception) {
-                "DefaultScope Not initialized ${e.message ?: "Unknown error"}"
-            }
-
         return """
             OneSignalDispatchers Status:
-            - IO Executor: $ioExecutorStatus
-            - Default Executor: $defaultExecutorStatus
-            - IO Scope: $ioScopeStatus
-            - Default Scope: $defaultScopeStatus
+            - IO Executor: ${executorStatus("ioExecutor") { ioExecutor.isShutdown }}
+            - Default Executor: ${executorStatus("defaultExecutor") { defaultExecutor.isShutdown }}
+            - SerialIO Executor: ${executorStatus("serialIOExecutor") { serialIOExecutor.isShutdown }}
+            - IO Scope: ${scopeStatus("IOScope") { IOScope.isActive }}
+            - Default Scope: ${scopeStatus("DefaultScope") { DefaultScope.isActive }}
+            - SerialIO Scope: ${scopeStatus("SerialIOScope") { SerialIOScope.isActive }}
         """.trimIndent()
     }
+
+    // Visible to tests so the failure branch (where `isShutdown()` itself throws — happens
+    // when the underlying executor's lazy initializer threw and re-throws on every access)
+    // can be exercised without forcing a real ThreadPoolExecutor to fail to construct, which
+    // is impractical to trigger in a unit test.
+    internal fun executorStatus(
+        name: String,
+        isShutdown: () -> Boolean,
+    ): String =
+        try {
+            if (isShutdown()) "Shutdown" else "Active"
+        } catch (e: Exception) {
+            "$name $NOT_INITIALIZED ${e.message ?: UNKNOWN_ERROR}"
+        }
+
+    internal fun scopeStatus(
+        name: String,
+        isActive: () -> Boolean,
+    ): String =
+        try {
+            if (isActive()) "Active" else "Cancelled"
+        } catch (e: Exception) {
+            "$name $NOT_INITIALIZED ${e.message ?: UNKNOWN_ERROR}"
+        }
+
+    private const val NOT_INITIALIZED = "Not initialized"
+    private const val UNKNOWN_ERROR = "Unknown error"
 }
diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/ThreadUtils.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/ThreadUtils.kt
index 92e3585386..8349e97851 100644
--- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/ThreadUtils.kt
+++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/common/threading/ThreadUtils.kt
@@ -98,6 +98,61 @@ fun suspendifyOnDefault(block: suspend () -> Unit) {
     suspendifyWithCompletion(useIO = false, block = block, onComplete = null)
 }
 
+/**
+ * Allows a non suspending function to create a scope that runs on the dedicated
+ * single-thread serial IO dispatcher. Tasks submitted from any thread run one-at-a-time
+ * in submission order, so callers do not need their own synchronization to preserve event
+ * ordering. This is the right entry point for lifecycle event handlers that do per-event
+ * work (timing, analytics, scheduler state) and must observe events in the order the
+ * Android framework fired them.
+ *
+ * Always routes through [OneSignalDispatchers.launchOnSerialIO] regardless of
+ * [ThreadingMode.useBackgroundThreading]: the serial ordering guarantee is the whole
+ * point of this helper, and the dedicated single thread carries none of the resource
+ * concerns that motivated the background-threading FF in the first place.
+ *
+ * Callers should capture any time-sensitive state (timestamps, "current" snapshots) on
+ * the caller's thread before invoking this — the work itself may run a few hundred ms
+ * later under load.
+ *
+ * @param block The suspending code to execute
+ */
+fun suspendifyOnSerialIO(block: suspend () -> Unit) {
+    OneSignalDispatchers.launchOnSerialIO {
+        try {
+            block()
+        } catch (e: Exception) {
+            Logging.error("Exception in suspendifyOnSerialIO", e)
+        }
+    }
+}
+
+/**
+ * Conditionally dispatches [block] to the serial IO thread when the SDK_BACKGROUND_THREADING
+ * remote feature flag is enabled, otherwise runs [block] inline on the calling thread.
+ *
+ * Used to gate incremental rollouts of off-main lifecycle work where we want to A/B compare
+ * the offloaded behavior against the previous inline behavior in production. Customers without
+ * the flag see the original code path; customers with the flag see the offloaded one.
+ *
+ * Differs from [suspendifyOnSerialIO] (which always routes through the serial dispatcher
+ * because its ordering guarantee is the whole point) and from [suspendifyOnIO] (which always
+ * dispatches off-main, just to a different pool depending on the flag).
+ *
+ * The block is non-suspending because the FF-off branch executes on the caller's thread —
+ * any suspending work would force a [kotlinx.coroutines.runBlocking] there, defeating the
+ * purpose of the comparison.
+ *
+ * @param block Non-suspending work to run inline (FF off) or on the serial IO thread (FF on).
+ */
+fun runOnSerialIOIfBackgroundThreading(block: () -> Unit) {
+    if (ThreadingMode.useBackgroundThreading) {
+        suspendifyOnSerialIO { block() }
+    } else {
+        block()
+    }
+}
+
 /**
  * Modern utility for executing suspending code with completion callback.
  * Uses OneSignal's centralized thread management for better resource control.
diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/background/impl/BackgroundManager.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/background/impl/BackgroundManager.kt
index 01c6c81934..c875912278 100644
--- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/background/impl/BackgroundManager.kt
+++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/background/impl/BackgroundManager.kt
@@ -32,6 +32,8 @@ import android.content.ComponentName
 import android.content.Context
 import android.content.pm.PackageManager
 import androidx.core.content.ContextCompat
+import com.onesignal.common.threading.ThreadingMode
+import com.onesignal.common.threading.suspendifyOnSerialIO
 import com.onesignal.core.internal.application.IApplicationLifecycleHandler
 import com.onesignal.core.internal.application.IApplicationService
 import com.onesignal.core.internal.background.IBackgroundManager
@@ -70,12 +72,42 @@ internal class BackgroundManager(
         _applicationService.addApplicationLifecycleHandler(this)
     }
 
+    // JobScheduler.cancel / .schedule are synchronous Binder transactions to
+    // system_server that can block the calling thread for many seconds on some
+    // devices (notably Xiaomi/MIUI under power-save). Lifecycle callbacks fire
+    // on the main thread, so running these inline produces the SDK-4505 ANR.
+    //
+    // The offload is gated on ThreadingMode.useBackgroundThreading (remote FF
+    // `sdk_background_threading`, latched at app startup) so the fix can be
+    // rolled out gradually and an A/B comparison against the legacy inline
+    // path is possible:
+    //
+    //   FF on  -> suspendifyOnSerialIO { ... }
+    //               Routes through OneSignalDispatchers.SerialIO, a dedicated
+    //               single-thread executor. Submission order on the main thread
+    //               equals execution order, so a rapid `unfocused -> focused`
+    //               burst can't reorder a stale `schedule` past a fresh
+    //               `cancel` (which a multi-threaded pool can). Required as
+    //               soon as these handlers grow per-event work (session
+    //               timing, analytics, focus counters) that can't be
+    //               reconciled from a single state snapshot.
+    //
+    //   FF off -> legacy inline path. Retains the ANR-prone behavior for the
+    //               control cohort.
     override fun onFocus(firedOnSubscribe: Boolean) {
-        cancelSyncTask()
+        if (ThreadingMode.useBackgroundThreading) {
+            suspendifyOnSerialIO { cancelSyncTask() }
+        } else {
+            cancelSyncTask()
+        }
     }
 
     override fun onUnfocused() {
-        scheduleBackground()
+        if (ThreadingMode.useBackgroundThreading) {
+            suspendifyOnSerialIO { scheduleBackground() }
+        } else {
+            scheduleBackground()
+        }
     }
 
     private fun scheduleBackground() {
diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/config/impl/FeatureFlagsRefreshService.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/config/impl/FeatureFlagsRefreshService.kt
index 40407f821f..304e43db27 100644
--- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/config/impl/FeatureFlagsRefreshService.kt
+++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/internal/config/impl/FeatureFlagsRefreshService.kt
@@ -4,6 +4,7 @@ import com.onesignal.common.modeling.ISingletonModelStoreChangeHandler
 import com.onesignal.common.modeling.ModelChangeTags
 import com.onesignal.common.modeling.ModelChangedArgs
 import com.onesignal.common.threading.OneSignalDispatchers
+import com.onesignal.common.threading.runOnSerialIOIfBackgroundThreading
 import com.onesignal.core.internal.application.IApplicationLifecycleHandler
 import com.onesignal.core.internal.application.IApplicationService
 import com.onesignal.core.internal.backend.IFeatureFlagsBackendService
@@ -69,15 +70,30 @@ internal class FeatureFlagsRefreshService(
         // Foreground-at-subscribe is handled by [IApplicationService.addApplicationLifecycleHandler] (fires onFocus).
     }
 
+    // restartForegroundPolling calls OneSignalDispatchers.launchOnIO, which on first cold use
+    // pays the executor + dispatcher + scope construction cost on the caller's thread. Since
+    // onFocus is delivered by ApplicationService.handleFocus on the main thread, that cost
+    // showed up as multi-second main-thread blocks under sdk_background_threading (SDK-4507).
+    //
+    // Gated on SDK_BACKGROUND_THREADING via runOnSerialIOIfBackgroundThreading so the FF-off
+    // cohort retains the original inline semantics (and the existing synchronous-onFocus tests
+    // observe the polling-job swap immediately). restartForegroundPolling itself takes the
+    // `synchronized(this)` lock that onModelUpdated / onModelReplaced share, so submission
+    // order preservation by the serial dispatcher is also a nice-to-have for back-to-back
+    // focus events.
     override fun onFocus(firedOnSubscribe: Boolean) {
-        restartForegroundPolling()
+        runOnSerialIOIfBackgroundThreading { restartForegroundPolling() }
     }
 
     override fun onUnfocused() {
-        synchronized(this) {
-            pollJob?.cancel()
-            pollJob = null
-            pollingAppId = null
+        runOnSerialIOIfBackgroundThreading {
+            // Qualify `this` so we lock on the FeatureFlagsRefreshService instance rather than
+            // the (no-receiver) lambda; same monitor restartForegroundPolling acquires.
+            synchronized(this@FeatureFlagsRefreshService) {
+                pollJob?.cancel()
+                pollJob = null
+                pollingAppId = null
+            }
         }
     }
 
diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/services/SyncJobService.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/services/SyncJobService.kt
index 055b269334..ff36ae649a 100644
--- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/services/SyncJobService.kt
+++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/core/services/SyncJobService.kt
@@ -29,12 +29,22 @@ package com.onesignal.core.services
 import android.app.job.JobParameters
 import android.app.job.JobService
 import com.onesignal.OneSignal
+import com.onesignal.common.threading.OneSignalDispatchers
 import com.onesignal.common.threading.suspendifyOnIO
 import com.onesignal.core.internal.background.IBackgroundManager
 import com.onesignal.debug.internal.logging.Logging
 
 class SyncJobService : JobService() {
     override fun onStartJob(jobParameters: JobParameters): Boolean {
+        // Android delivers JobService.onStartJob on the main thread. The suspendifyOnIO call
+        // below is the SDK's first IO-pool consumer on cold start in this process, and the
+        // executor + dispatcher + coroutine-scope lazy chain it triggers was producing multi-
+        // second main-thread blocks in production (SDK-4507). prewarm() shifts that cost to a
+        // short-lived daemon thread; idempotent, so it's harmless when initWithContext already
+        // ran. Must happen BEFORE suspendifyOnIO; calling it inside the dispatched block would
+        // be too late because the cold-init cost has already been paid on entry to the helper.
+        OneSignalDispatchers.prewarm()
+
         suspendifyOnIO {
             var reschedule = false
 
diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/internal/OneSignalImp.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/internal/OneSignalImp.kt
index c8de167481..8108e9ce84 100644
--- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/internal/OneSignalImp.kt
+++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/internal/OneSignalImp.kt
@@ -309,6 +309,14 @@ internal class OneSignalImp(
     ): Boolean {
         Logging.log(LogLevel.DEBUG, "Calling deprecated initWithContext(context: $context, appId: $appId)")
 
+        // Warm OneSignalDispatchers on a dedicated daemon thread so the first production caller
+        // of suspendifyOnIO / launchOnSerialIO doesn't pay the ThreadPoolExecutor + dispatcher +
+        // scope construction cost on the main thread. See SDK-4507; OTel showed that cost as
+        // 5–20s main-thread blocks at first foreground/background lifecycle event under
+        // sdk_background_threading. Calling prewarm() is idempotent, fire-and-forget, and safe
+        // even if a prior initWithContext attempt already started the prewarm.
+        OneSignalDispatchers.prewarm()
+
         synchronized(initLock) {
             if (initState.isSDKAccessible()) {
                 Logging.log(LogLevel.DEBUG, "initWithContext: SDK already initialized or in progress")
@@ -808,6 +816,12 @@ internal class OneSignalImp(
     ): Boolean {
         Logging.log(LogLevel.DEBUG, "initWithContext(context: $context, appId: $appId)")
 
+        // Same SDK-4507 warm-up as the synchronous variant. Reaching this entry point on the
+        // main thread (e.g. SyncJobService.onStartJob -> suspendifyOnIO -> initWithContext(context))
+        // pays the cold-init cost on the dispatcher used to enter [withContext] below, so warm
+        // OneSignalDispatchers on a background thread before we touch [runtimeIoDispatcher].
+        OneSignalDispatchers.prewarm()
+
         // Use IO dispatcher for initialization to prevent ANRs and optimize for I/O operations
         return withContext(runtimeIoDispatcher) {
             val shouldRunInit: Boolean
diff --git a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/impl/SessionService.kt b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/impl/SessionService.kt
index 1997d486b7..30b00b2ecd 100644
--- a/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/impl/SessionService.kt
+++ b/OneSignalSDK/onesignal/core/src/main/java/com/onesignal/session/internal/session/impl/SessionService.kt
@@ -1,6 +1,7 @@
 package com.onesignal.session.internal.session.impl
 
 import com.onesignal.common.events.EventProducer
+import com.onesignal.common.threading.runOnSerialIOIfBackgroundThreading
 import com.onesignal.core.internal.application.IApplicationLifecycleHandler
 import com.onesignal.core.internal.application.IApplicationService
 import com.onesignal.core.internal.background.IBackgroundService
@@ -101,8 +102,35 @@ internal class SessionService(
      * In this case, the app is already foregrounded, so this method is fired immediately on subscribing
      * to the IApplicationService. Listeners of this service will not subscribe in time to capture
      * the `onSessionStarted()` callback here, so fire it when they themselves subscribe.
+     *
+     * SDK-4508: Production OTel shows this handler fired from `ApplicationService.handleFocus` on
+     * the main thread blocking for many seconds when the cold-init dispatcher / executor lazy
+     * chain inside [com.onesignal.common.threading.OneSignalDispatchers] runs as a downstream
+     * effect of `sessionLifeCycleNotifier.fire { onSessionStarted/Active }`. The subscribed
+     * handlers (operation repo, IAM trigger eval, etc.) hand work off to those pools, and the
+     * first of them to touch the chain pays the construction cost on whichever thread is
+     * delivering the event.
+     *
+     * We capture timing-sensitive state on the caller's thread (so `session.startTime` /
+     * `session.focusTime` reflect the moment Android delivered the event, not whenever the
+     * serial dispatcher gets around to running our block) and then hand the rest of the work
+     * to [runOnSerialIOIfBackgroundThreading]. FF off retains the original inline semantics for
+     * the rollout control cohort and the existing synchronous-onFocus tests.
      */
     override fun onFocus(firedOnSubscribe: Boolean) {
+        // Capture timing state BEFORE handing off so the session timestamps reflect main-thread
+        // event arrival, not whenever the serial dispatcher runs the block. Same pattern the
+        // KDoc on `suspendifyOnSerialIO` recommends for time-sensitive lifecycle work.
+        val focusTimeMs = _time.currentTimeMillis
+        runOnSerialIOIfBackgroundThreading {
+            handleOnFocus(firedOnSubscribe, focusTimeMs)
+        }
+    }
+
+    private fun handleOnFocus(
+        firedOnSubscribe: Boolean,
+        focusTimeMs: Long,
+    ) {
         Logging.log(LogLevel.DEBUG, "SessionService.onFocus() - fired from start: $firedOnSubscribe")
 
         val session = this.session
@@ -121,7 +149,7 @@ internal class SessionService(
             // As the old session was made inactive, we need to create a new session
             shouldFireOnSubscribe = firedOnSubscribe
             session.sessionId = UUID.randomUUID().toString()
-            session.startTime = _time.currentTimeMillis
+            session.startTime = focusTimeMs
             session.focusTime = session.startTime
             session.isValid = true
             Logging.debug("SessionService: New session started at ${session.startTime}")
@@ -129,19 +157,28 @@ internal class SessionService(
         } else {
             // existing session: just remember the focus time so we can calculate the active time
             // when onUnfocused is called.
-            session.focusTime = _time.currentTimeMillis
+            session.focusTime = focusTimeMs
             sessionLifeCycleNotifier.fire { it.onSessionActive() }
         }
     }
 
     override fun onUnfocused() {
+        // Same pattern as onFocus: capture the timing on the caller's thread so activeDuration
+        // accounting is unaffected by serial-dispatcher latency, then offload.
+        val unfocusTimeMs = _time.currentTimeMillis
+        runOnSerialIOIfBackgroundThreading {
+            handleOnUnfocused(unfocusTimeMs)
+        }
+    }
+
+    private fun handleOnUnfocused(unfocusTimeMs: Long) {
         val session = this.session
         if (session == null) {
             Logging.warn("SessionService.onUnfocused called before bootstrap; ignoring.")
             return
         }
         // capture the amount of time the app was focused
-        val dt = _time.currentTimeMillis - session.focusTime
+        val dt = unfocusTimeMs - session.focusTime
         session.activeDuration += dt
         Logging.log(LogLevel.DEBUG, "SessionService.onUnfocused adding time $dt for total: ${session.activeDuration}")
     }
diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/OneSignalDispatchersTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/OneSignalDispatchersTests.kt
index 72dc5e2b91..121a195dc6 100644
--- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/OneSignalDispatchersTests.kt
+++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/OneSignalDispatchersTests.kt
@@ -82,8 +82,118 @@ class OneSignalDispatchersTests : FunSpec({
         status shouldContain "OneSignalDispatchers Status:"
         status shouldContain "IO Executor: Active"
         status shouldContain "Default Executor: Active"
+        status shouldContain "SerialIO Executor: Active"
         status shouldContain "IO Scope: Active"
         status shouldContain "Default Scope: Active"
+        status shouldContain "SerialIO Scope: Active"
+    }
+
+    test("getPerformanceMetrics should include SerialIO queue and total completed task counters") {
+        // Trigger lazy init of the SerialIO executor before asking for metrics so its queue
+        // line resolves to a concrete value instead of the "n/a" fallback path.
+        OneSignalDispatchers.SerialIO shouldNotBe null
+
+        val metrics = OneSignalDispatchers.getPerformanceMetrics()
+
+        metrics shouldContain "OneSignalDispatchers Performance Metrics:"
+        metrics shouldContain "SerialIO Queue:"
+        metrics shouldContain "Total completed tasks:"
+    }
+
+    test("SerialIO dispatcher executes work on a background thread") {
+        val callerThreadId = Thread.currentThread().id
+        var serialThreadId: Long? = null
+
+        runBlocking {
+            withContext(OneSignalDispatchers.SerialIO) {
+                serialThreadId = Thread.currentThread().id
+            }
+        }
+
+        serialThreadId shouldNotBe null
+        serialThreadId shouldNotBe callerThreadId
+    }
+
+    test("launchOnSerialIO runs tasks on a single thread in submission order") {
+        // SerialIO's contract: submission order on the caller thread == execution order on
+        // the worker thread. We submit N tasks with a small sleep so they queue up, then
+        // assert the recorded order matches submission order and that all observations
+        // came from one thread.
+        val taskCount = 5
+        val observedOrder = mutableListOf<Int>()
+        val observedThreads = mutableSetOf<Long>()
+        val latch = CountDownLatch(taskCount)
+
+        repeat(taskCount) { i ->
+            OneSignalDispatchers.launchOnSerialIO {
+                Thread.sleep(5)
+                synchronized(observedOrder) {
+                    observedOrder.add(i)
+                    observedThreads.add(Thread.currentThread().id)
+                }
+                latch.countDown()
+            }
+        }
+
+        latch.await()
+        observedOrder shouldBe (0 until taskCount).toList()
+        observedThreads.size shouldBe 1
+    }
+
+    test("executorStatus returns 'Active' / 'Shutdown' on the happy path and the Not initialized message when the underlying check throws") {
+        // Happy paths (Shutdown / Active) are exercised indirectly via getStatus(); this
+        // case pins down the defensive catch branch, which fires when the underlying
+        // executor's lazy initializer is in a failed state (e.g. JVM-level
+        // Executors.newSingleThreadExecutor refused to construct) and every isShutdown
+        // access re-throws. Without this, the catch is unreachable from unit tests because
+        // ThreadPoolExecutor.isShutdown does not normally throw.
+        OneSignalDispatchers.executorStatus("ioExecutor") { false } shouldBe "Active"
+        OneSignalDispatchers.executorStatus("ioExecutor") { true } shouldBe "Shutdown"
+        OneSignalDispatchers.executorStatus("ioExecutor") {
+            throw RuntimeException("init failure")
+        } shouldBe "ioExecutor Not initialized init failure"
+        OneSignalDispatchers.executorStatus("ioExecutor") {
+            throw RuntimeException()
+        } shouldBe "ioExecutor Not initialized Unknown error"
+    }
+
+    test("scopeStatus returns 'Active' / 'Cancelled' on the happy path and the Not initialized message when the underlying check throws") {
+        OneSignalDispatchers.scopeStatus("IOScope") { true } shouldBe "Active"
+        OneSignalDispatchers.scopeStatus("IOScope") { false } shouldBe "Cancelled"
+        OneSignalDispatchers.scopeStatus("IOScope") {
+            throw RuntimeException("cancelled supervisor")
+        } shouldBe "IOScope Not initialized cancelled supervisor"
+        OneSignalDispatchers.scopeStatus("IOScope") {
+            throw RuntimeException()
+        } shouldBe "IOScope Not initialized Unknown error"
+    }
+
+    test("exceptions in a SerialIO task do not stop subsequent tasks from running") {
+        // Mirrors the parallel "exceptions in one task should not affect others" case for
+        // launchOnIO. A thrown exception in one serial task must not poison the dispatcher
+        // for the rest of the queue.
+        val latch = CountDownLatch(3)
+        val successCount = AtomicInteger(0)
+        val errorCount = AtomicInteger(0)
+
+        repeat(3) { i ->
+            OneSignalDispatchers.launchOnSerialIO {
+                try {
+                    if (i == 1) {
+                        throw RuntimeException("Test error")
+                    }
+                    successCount.incrementAndGet()
+                } catch (e: Exception) {
+                    errorCount.incrementAndGet()
+                } finally {
+                    latch.countDown()
+                }
+            }
+        }
+
+        latch.await()
+        successCount.get() shouldBe 2
+        errorCount.get() shouldBe 1
     }
 
     test("dispatchers should handle concurrent operations") {
@@ -171,4 +281,61 @@ class OneSignalDispatchersTests : FunSpec({
         successCount.get() shouldBe 4
         errorCount.get() shouldBe 1
     }
+
+    test("prewarm returns immediately and warms IO / Default / SerialIO dispatchers on a background thread") {
+        // SDK-4507: regression coverage for the cold-init main-thread block. prewarm() must
+        // (a) return on the caller's thread without ever doing the executor / dispatcher /
+        // scope construction work inline, and (b) leave all three dispatchers + scopes in the
+        // "Active" state once the dedicated daemon thread finishes its empty launches.
+        OneSignalDispatchers.resetPrewarmForTest()
+        val callerThreadId = Thread.currentThread().id
+
+        // Call from the test thread (which stands in for the main thread under production
+        // usage). The call must return microseconds-fast; we don't assert wall-clock latency,
+        // just that the heavy work didn't happen on this thread.
+        OneSignalDispatchers.prewarm()
+
+        // Resolve the prewarm thread by name from the JVM's thread set; its name is set by
+        // the prewarm() impl. We `join()` on it so the subsequent status assertions don't
+        // race a still-running prewarm thread.
+        val prewarmThread =
+            Thread.getAllStackTraces().keys.firstOrNull { it.name == "OneSignal-prewarm" }
+        prewarmThread?.join(2_000)
+        // After prewarm has finished, getStatus must report all three executors and scopes
+        // as Active. If the prewarm thread itself failed it would be a no-op for getStatus
+        // because the lazy chain wouldn't have run; this assertion proves both ends of the
+        // contract (heavy work was done, and it ran on the prewarm thread, not the caller).
+        val status = OneSignalDispatchers.getStatus()
+        status shouldContain "IO Executor: Active"
+        status shouldContain "Default Executor: Active"
+        status shouldContain "SerialIO Executor: Active"
+        status shouldContain "IO Scope: Active"
+        status shouldContain "Default Scope: Active"
+        status shouldContain "SerialIO Scope: Active"
+
+        // Sanity: the prewarm thread was a separate thread, not the test thread.
+        prewarmThread?.id shouldNotBe callerThreadId
+    }
+
+    test("prewarm is idempotent: a second call is a no-op and does not spawn a second prewarm thread") {
+        // The first prewarm() may have already run in earlier tests (or in the previous test
+        // above). Reset the latch so we get a deterministic "first call" here, then verify
+        // that the second call does NOT spawn another OneSignal-prewarm thread.
+        OneSignalDispatchers.resetPrewarmForTest()
+
+        OneSignalDispatchers.prewarm()
+        val firstPrewarmThread =
+            Thread.getAllStackTraces().keys.firstOrNull { it.name == "OneSignal-prewarm" }
+        firstPrewarmThread?.join(2_000)
+
+        // Snapshot any straggling "OneSignal-prewarm" threads -- there should be at most one
+        // (the one above, possibly still in TERMINATED state in the JVM's thread set briefly).
+        val countBefore = Thread.getAllStackTraces().keys.count { it.name == "OneSignal-prewarm" }
+
+        // Second call must be a no-op. No new prewarm thread, no exception.
+        OneSignalDispatchers.prewarm()
+        val countAfter = Thread.getAllStackTraces().keys.count { it.name == "OneSignal-prewarm" }
+
+        countAfter shouldBe countBefore
+    }
 })
diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/ThreadUtilsFeatureFlagTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/ThreadUtilsFeatureFlagTests.kt
index 3e529f1c3c..cbdbc7b935 100644
--- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/ThreadUtilsFeatureFlagTests.kt
+++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/common/threading/ThreadUtilsFeatureFlagTests.kt
@@ -10,6 +10,8 @@ import io.mockk.verify
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.runBlocking
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
 
 class ThreadUtilsFeatureFlagTests : FunSpec({
     beforeEach {
@@ -82,4 +84,71 @@ class ThreadUtilsFeatureFlagTests : FunSpec({
         completed.isCompleted shouldBe true
         verify(exactly = 0) { OneSignalDispatchers.launchOnDefault(any<suspend () -> Unit>()) }
     }
+
+    test("suspendifyOnSerialIO always routes through OneSignalDispatchers.launchOnSerialIO regardless of BACKGROUND_THREADING") {
+        // suspendifyOnSerialIO intentionally ignores the FF: the serial ordering guarantee
+        // is the whole point of this entry point, and the single low-priority daemon thread
+        // carries none of the resource concerns the FF gates. Exercise both FF positions in
+        // one test to lock in that contract.
+        listOf(false, true).forEach { ffOn ->
+            ThreadingMode.useBackgroundThreading = ffOn
+            mockkObject(OneSignalDispatchers)
+            every { OneSignalDispatchers.launchOnSerialIO(any<suspend () -> Unit>()) } returns mockk<Job>(relaxed = true)
+
+            suspendifyOnSerialIO { }
+
+            verify(exactly = 1) { OneSignalDispatchers.launchOnSerialIO(any<suspend () -> Unit>()) }
+            unmockkObject(OneSignalDispatchers)
+        }
+    }
+
+    test("suspendifyOnSerialIO swallows exceptions thrown inside the block") {
+        // Production contract: any exception in the dispatched block is logged and absorbed
+        // rather than propagated to the SerialIO thread, so a single misbehaving caller
+        // can't kill the dispatcher for the rest of the SDK.
+        var ranBlock = false
+
+        suspendifyOnSerialIO {
+            ranBlock = true
+            throw RuntimeException("intentional")
+        }
+
+        // Drain the SerialIO worker: submit a follow-up task and wait for it. If exception
+        // handling worked the block above ran and the follow-up runs too.
+        val latch = CountDownLatch(1)
+        suspendifyOnSerialIO { latch.countDown() }
+        latch.await(2, TimeUnit.SECONDS) shouldBe true
+        ranBlock shouldBe true
+    }
+
+    test("runOnSerialIOIfBackgroundThreading routes through launchOnSerialIO when BACKGROUND_THREADING is on") {
+        // Given
+        ThreadingMode.useBackgroundThreading = true
+        mockkObject(OneSignalDispatchers)
+        every { OneSignalDispatchers.launchOnSerialIO(any<suspend () -> Unit>()) } returns mockk<Job>(relaxed = true)
+        var ranInline = false
+
+        // When
+        runOnSerialIOIfBackgroundThreading { ranInline = true }
+
+        // Then
+        ranInline shouldBe false
+        verify(exactly = 1) { OneSignalDispatchers.launchOnSerialIO(any<suspend () -> Unit>()) }
+    }
+
+    test("runOnSerialIOIfBackgroundThreading runs inline on caller thread when BACKGROUND_THREADING is off") {
+        // Given
+        ThreadingMode.useBackgroundThreading = false
+        mockkObject(OneSignalDispatchers)
+        every { OneSignalDispatchers.launchOnSerialIO(any<suspend () -> Unit>()) } returns mockk<Job>(relaxed = true)
+        val callerThread = Thread.currentThread()
+        var ranOnThread: Thread? = null
+
+        // When
+        runOnSerialIOIfBackgroundThreading { ranOnThread = Thread.currentThread() }
+
+        // Then
+        ranOnThread shouldBe callerThread
+        verify(exactly = 0) { OneSignalDispatchers.launchOnSerialIO(any<suspend () -> Unit>()) }
+    }
 })
diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/background/impl/BackgroundManagerTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/background/impl/BackgroundManagerTests.kt
new file mode 100644
index 0000000000..7fbf570b1d
--- /dev/null
+++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/background/impl/BackgroundManagerTests.kt
@@ -0,0 +1,149 @@
+package com.onesignal.core.internal.background.impl
+
+import android.app.job.JobScheduler
+import android.content.Context
+import com.onesignal.common.threading.OneSignalDispatchers
+import com.onesignal.common.threading.ThreadingMode
+import com.onesignal.core.internal.application.IApplicationService
+import com.onesignal.debug.LogLevel
+import com.onesignal.debug.internal.logging.Logging
+import com.onesignal.mocks.MockHelper
+import io.kotest.core.spec.style.FunSpec
+import io.mockk.every
+import io.mockk.mockk
+import io.mockk.mockkObject
+import io.mockk.unmockkObject
+import io.mockk.verify
+import io.mockk.verifyOrder
+import kotlinx.coroutines.Job
+
+/**
+ * Regression coverage for SDK-4505. Asserts the two-pronged behavior of
+ * BackgroundManager.onFocus / onUnfocused:
+ *
+ *   FF on  (sdk_background_threading enabled)
+ *     -> route through OneSignalDispatchers.launchOnSerialIO so the
+ *        JobScheduler Binder call doesn't run inline on the main thread
+ *        (the ANR fix), and rapid bursts stay in submission order (the
+ *        serial-dispatcher refinement).
+ *
+ *   FF off
+ *     -> legacy inline path. cancelSyncTask / scheduleBackground execute
+ *        on the calling thread; no dispatcher is involved. This is the
+ *        control cohort for the gated rollout.
+ */
+class BackgroundManagerTests : FunSpec({
+
+    fun applicationServiceWithStubbedJobScheduler(): IApplicationService {
+        val appService = MockHelper.applicationService()
+        val context = mockk<Context>(relaxed = true)
+        val jobScheduler = mockk<JobScheduler>(relaxed = true)
+        every { appService.appContext } returns context
+        every { context.getSystemService(Context.JOB_SCHEDULER_SERVICE) } returns jobScheduler
+        every { jobScheduler.allPendingJobs } returns emptyList()
+        return appService
+    }
+
+    beforeEach {
+        Logging.logLevel = LogLevel.NONE
+        ThreadingMode.useBackgroundThreading = false
+    }
+
+    afterEach {
+        unmockkObject(OneSignalDispatchers)
+        ThreadingMode.useBackgroundThreading = false
+    }
+
+    test("FF on: onFocus routes through OneSignalDispatchers.launchOnSerialIO") {
+        ThreadingMode.useBackgroundThreading = true
+        mockkObject(OneSignalDispatchers)
+        every { OneSignalDispatchers.launchOnSerialIO(any<suspend () -> Unit>()) } returns mockk<Job>(relaxed = true)
+
+        val backgroundManager =
+            BackgroundManager(
+                MockHelper.applicationService(),
+                MockHelper.time(0L),
+                emptyList(),
+            )
+
+        backgroundManager.onFocus(firedOnSubscribe = false)
+
+        verify(exactly = 1) { OneSignalDispatchers.launchOnSerialIO(any<suspend () -> Unit>()) }
+    }
+
+    test("FF on: onUnfocused routes through OneSignalDispatchers.launchOnSerialIO") {
+        ThreadingMode.useBackgroundThreading = true
+        mockkObject(OneSignalDispatchers)
+        every { OneSignalDispatchers.launchOnSerialIO(any<suspend () -> Unit>()) } returns mockk<Job>(relaxed = true)
+
+        val backgroundManager =
+            BackgroundManager(
+                MockHelper.applicationService(),
+                MockHelper.time(0L),
+                emptyList(),
+            )
+
+        backgroundManager.onUnfocused()
+
+        verify(exactly = 1) { OneSignalDispatchers.launchOnSerialIO(any<suspend () -> Unit>()) }
+    }
+
+    test("FF on: rapid unfocus -> focus burst submits both events to the serial dispatcher in order") {
+        ThreadingMode.useBackgroundThreading = true
+        mockkObject(OneSignalDispatchers)
+        every { OneSignalDispatchers.launchOnSerialIO(any<suspend () -> Unit>()) } returns mockk<Job>(relaxed = true)
+
+        val backgroundManager =
+            BackgroundManager(
+                MockHelper.applicationService(),
+                MockHelper.time(0L),
+                emptyList(),
+            )
+
+        // Simulate the user backgrounding then immediately foregrounding the app on the
+        // main thread (the SDK-4505 reproducer). Both calls must route through the same
+        // serial dispatcher so the IO worker observes them in this submission order.
+        backgroundManager.onUnfocused()
+        backgroundManager.onFocus(firedOnSubscribe = false)
+
+        verify(exactly = 2) { OneSignalDispatchers.launchOnSerialIO(any<suspend () -> Unit>()) }
+        verifyOrder {
+            OneSignalDispatchers.launchOnSerialIO(any<suspend () -> Unit>())
+            OneSignalDispatchers.launchOnSerialIO(any<suspend () -> Unit>())
+        }
+    }
+
+    test("FF off: onFocus runs inline and does NOT dispatch to the serial dispatcher") {
+        ThreadingMode.useBackgroundThreading = false
+        mockkObject(OneSignalDispatchers)
+        every { OneSignalDispatchers.launchOnSerialIO(any<suspend () -> Unit>()) } returns mockk<Job>(relaxed = true)
+
+        val backgroundManager =
+            BackgroundManager(
+                applicationServiceWithStubbedJobScheduler(),
+                MockHelper.time(0L),
+                emptyList(),
+            )
+
+        backgroundManager.onFocus(firedOnSubscribe = false)
+
+        verify(exactly = 0) { OneSignalDispatchers.launchOnSerialIO(any<suspend () -> Unit>()) }
+    }
+
+    test("FF off: onUnfocused runs inline and does NOT dispatch to the serial dispatcher") {
+        ThreadingMode.useBackgroundThreading = false
+        mockkObject(OneSignalDispatchers)
+        every { OneSignalDispatchers.launchOnSerialIO(any<suspend () -> Unit>()) } returns mockk<Job>(relaxed = true)
+
+        val backgroundManager =
+            BackgroundManager(
+                applicationServiceWithStubbedJobScheduler(),
+                MockHelper.time(0L),
+                emptyList(),
+            )
+
+        backgroundManager.onUnfocused()
+
+        verify(exactly = 0) { OneSignalDispatchers.launchOnSerialIO(any<suspend () -> Unit>()) }
+    }
+})
diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/config/impl/FeatureFlagsRefreshServiceTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/config/impl/FeatureFlagsRefreshServiceTests.kt
index 5ceb40e77e..54175bbb1e 100644
--- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/config/impl/FeatureFlagsRefreshServiceTests.kt
+++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/core/internal/config/impl/FeatureFlagsRefreshServiceTests.kt
@@ -1,6 +1,8 @@
 package com.onesignal.core.internal.config.impl
 
 import com.onesignal.common.modeling.ModelChangeTags
+import com.onesignal.common.threading.OneSignalDispatchers
+import com.onesignal.common.threading.runOnSerialIOIfBackgroundThreading
 import com.onesignal.core.internal.application.IApplicationLifecycleHandler
 import com.onesignal.core.internal.application.IApplicationService
 import com.onesignal.core.internal.backend.IFeatureFlagsBackendService
@@ -17,7 +19,10 @@ import io.mockk.coVerify
 import io.mockk.every
 import io.mockk.just
 import io.mockk.mockk
+import io.mockk.mockkStatic
 import io.mockk.slot
+import io.mockk.unmockkStatic
+import io.mockk.verify
 
 /**
  * Regression coverage for the duplicate Turbine feature-flags fetch at SDK startup.
@@ -187,4 +192,44 @@ class FeatureFlagsRefreshServiceTests : FunSpec({
         awaitIO()
         fetchCount() shouldBe 2
     }
+
+    test("onFocus / onUnfocused route through runOnSerialIOIfBackgroundThreading (SDK-4507)") {
+        // SDK-4507: the lifecycle handlers run on the main thread via
+        // ApplicationService.handleFocus -> applicationLifecycleNotifier.fire. The body of
+        // restartForegroundPolling calls OneSignalDispatchers.launchOnIO, which on first cold
+        // use pays the executor + dispatcher + scope construction cost on the calling thread.
+        // The fix wraps both handlers in runOnSerialIOIfBackgroundThreading; this test pins
+        // down the dispatch contract.
+        //
+        // Reset the cumulative call counter on ThreadUtilsKt (IOMockHelper installs the static
+        // mock at spec-level, so calls from earlier tests in this spec would otherwise count
+        // against our `verify(exactly = ...)` assertion). Unmock + remock is the cheapest way
+        // to drop the recorded calls; we then re-install the inline-run answer IOMockHelper
+        // provided so the wrapped block still executes and onFocus/onUnfocused keep their
+        // side effects.
+        unmockkStatic("com.onesignal.common.threading.ThreadUtilsKt")
+        mockkStatic("com.onesignal.common.threading.ThreadUtilsKt")
+        every { runOnSerialIOIfBackgroundThreading(any<() -> Unit>()) } answers {
+            firstArg<() -> Unit>().invoke()
+        }
+        every { OneSignalDispatchers.launchOnIO(any<suspend () -> Unit>()) } returns mockk(relaxed = true)
+
+        val model = ConfigModel().apply { appId = "appId-1" }
+        val store = mockConfigStore(model)
+        val (backend, _) = mockBackend()
+        // start: [true, false] (loop iter1=true, iter2=false break) + onFocus restart loop: [true, false].
+        val app = foregroundedAppService(true, false, true, false)
+
+        val service = FeatureFlagsRefreshService(app, store, backend).apply { refreshIntervalMs = 0L }
+        service.start()
+        awaitIO()
+
+        // start() fires onFocus(true) via the addApplicationLifecycleHandler mock, so we
+        // already have 1 invocation from initial focus. Direct onFocus / onUnfocused calls
+        // bump the counter to 3 total.
+        service.onFocus(firedOnSubscribe = false)
+        service.onUnfocused()
+
+        verify(exactly = 3) { runOnSerialIOIfBackgroundThreading(any<() -> Unit>()) }
+    }
 })
diff --git a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/SessionServiceTests.kt b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/SessionServiceTests.kt
index 3e275bbb79..23087bbc32 100644
--- a/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/SessionServiceTests.kt
+++ b/OneSignalSDK/onesignal/core/src/test/java/com/onesignal/session/internal/session/SessionServiceTests.kt
@@ -1,11 +1,23 @@
 package com.onesignal.session.internal.session
 
+import com.onesignal.common.threading.OneSignalDispatchers
+import com.onesignal.common.threading.runOnSerialIOIfBackgroundThreading
 import com.onesignal.mocks.MockHelper
 import com.onesignal.session.internal.session.impl.SessionService
 import io.kotest.core.spec.style.FunSpec
 import io.kotest.matchers.shouldBe
+import io.mockk.every
+import io.mockk.just
+import io.mockk.mockk
+import io.mockk.mockkObject
+import io.mockk.mockkStatic
+import io.mockk.runs
 import io.mockk.spyk
+import io.mockk.unmockkObject
+import io.mockk.unmockkStatic
 import io.mockk.verify
+import io.mockk.verifyOrder
+import kotlinx.coroutines.Job
 
 // Mocks used by every test in this file
 private class Mocks {
@@ -162,4 +174,99 @@ class SessionServiceTests : FunSpec({
         // Then
         verify(exactly = 0) { mocks.spyCallback.onSessionEnded(any()) }
     }
+
+    test("onFocus dispatches the session-mutation body through runOnSerialIOIfBackgroundThreading (SDK-4508)") {
+        // SDK-4508: SessionService.onFocus runs on the main thread via
+        // ApplicationService.handleFocus -> applicationLifecycleNotifier.fire. Its body fires
+        // session lifecycle handlers (operation repo, IAM trigger eval, etc.) which can in turn
+        // touch OneSignalDispatchers' cold-init chain. The fix wraps the body in
+        // runOnSerialIOIfBackgroundThreading; this test pins down the dispatch contract.
+        //
+        // Stub the helper as a pass-through so the underlying state mutations still happen
+        // (`startTime`, `focusTime`, lifecycle-handler fires) and the existing assertions
+        // about session state remain meaningful.
+        val threadUtilsPath = "com.onesignal.common.threading.ThreadUtilsKt"
+        mockkStatic(threadUtilsPath)
+        mockkObject(OneSignalDispatchers)
+        every { runOnSerialIOIfBackgroundThreading(any<() -> Unit>()) } answers {
+            firstArg<() -> Unit>().invoke()
+        }
+        every { OneSignalDispatchers.launchOnSerialIO(any<suspend () -> Unit>()) } returns mockk<Job>(relaxed = true)
+
+        try {
+            val mocks = Mocks()
+            val sessionService = mocks.sessionService
+            sessionService.bootstrap()
+            sessionService.start()
+            mocks.sessionModelStore { it.isValid = false }
+
+            sessionService.onFocus(firedOnSubscribe = false)
+
+            verify(exactly = 1) { runOnSerialIOIfBackgroundThreading(any<() -> Unit>()) }
+        } finally {
+            unmockkObject(OneSignalDispatchers)
+            unmockkStatic(threadUtilsPath)
+        }
+    }
+
+    test("onUnfocused dispatches the activeDuration update through runOnSerialIOIfBackgroundThreading (SDK-4508)") {
+        val threadUtilsPath = "com.onesignal.common.threading.ThreadUtilsKt"
+        mockkStatic(threadUtilsPath)
+        mockkObject(OneSignalDispatchers)
+        every { runOnSerialIOIfBackgroundThreading(any<() -> Unit>()) } answers {
+            firstArg<() -> Unit>().invoke()
+        }
+        every { OneSignalDispatchers.launchOnSerialIO(any<suspend () -> Unit>()) } returns mockk<Job>(relaxed = true)
+
+        try {
+            val mocks = Mocks()
+            val sessionService = mocks.sessionService
+            sessionService.bootstrap()
+            sessionService.start()
+            mocks.sessionModelStore {
+                it.isValid = true
+                it.focusTime = 0L
+            }
+
+            sessionService.onUnfocused()
+
+            verify(exactly = 1) { runOnSerialIOIfBackgroundThreading(any<() -> Unit>()) }
+        } finally {
+            unmockkObject(OneSignalDispatchers)
+            unmockkStatic(threadUtilsPath)
+        }
+    }
+
+    test("rapid onUnfocused -> onFocus burst dispatches each event through the gated helper in submission order (SDK-4508)") {
+        // Mirrors the SDK-4505 BackgroundManager burst test. Real-world scenario: the user
+        // backgrounds then immediately re-foregrounds the app on the main thread. Both lifecycle
+        // events must route through the same gated helper in submission order so the serial IO
+        // worker sees focusTime / activeDuration mutations in main-thread arrival order. If they
+        // ever raced across the IO pool, activeDuration accounting could drift.
+        val threadUtilsPath = "com.onesignal.common.threading.ThreadUtilsKt"
+        mockkStatic(threadUtilsPath)
+        mockkObject(OneSignalDispatchers)
+        every { runOnSerialIOIfBackgroundThreading(any<() -> Unit>()) } just runs
+        every { OneSignalDispatchers.launchOnSerialIO(any<suspend () -> Unit>()) } returns mockk<Job>(relaxed = true)
+
+        try {
+            val mocks = Mocks()
+            val sessionService = mocks.sessionService
+            sessionService.bootstrap()
+            sessionService.start()
+            mocks.sessionModelStore { it.isValid = true }
+
+            sessionService.onUnfocused()
+            sessionService.onFocus(firedOnSubscribe = false)
+
+            verify(exactly = 2) { runOnSerialIOIfBackgroundThreading(any<() -> Unit>()) }
+            verifyOrder {
+                runOnSerialIOIfBackgroundThreading(any<() -> Unit>())
+                runOnSerialIOIfBackgroundThreading(any<() -> Unit>())
+            }
+        } finally {
+            unmockkObject(OneSignalDispatchers)
+            unmockkStatic(threadUtilsPath)
+        }
+    }
 })
diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/internal/NotificationsManager.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/internal/NotificationsManager.kt
index fd5578e480..ceeba46d3d 100644
--- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/internal/NotificationsManager.kt
+++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/internal/NotificationsManager.kt
@@ -2,6 +2,7 @@ package com.onesignal.notifications.internal
 
 import android.app.Activity
 import com.onesignal.common.events.EventProducer
+import com.onesignal.common.threading.runOnSerialIOIfBackgroundThreading
 import com.onesignal.common.threading.suspendifyOnIO
 import com.onesignal.core.internal.application.IApplicationLifecycleHandler
 import com.onesignal.core.internal.application.IApplicationService
@@ -58,8 +59,23 @@ internal class NotificationsManager(
         }
     }
 
+    // beginEnqueueingWork performs synchronous WorkManager DB I/O (and on first
+    // call lazily initializes WorkManager itself, opening/migrating its SQLite
+    // store). Lifecycle callbacks fire on the main thread, so on devices where
+    // that I/O is slow — heavy storage contention, low memory, OEM throttling —
+    // the call can block the main thread for many seconds and trip an ANR.
+    //
+    // Gated on SDK_BACKGROUND_THREADING via runOnSerialIOIfBackgroundThreading
+    // so we can A/B compare the offloaded behavior (FF on → serial IO dispatch)
+    // against the previous inline behavior (FF off → main thread) in production.
+    // The serial dispatcher is the same one BackgroundManager uses for its
+    // lifecycle JobScheduler calls; keeping both handlers on it preserves
+    // submission order on the main thread = execution order on the serial
+    // thread, and leaves room to add per-event work here later (focus counters,
+    // notification analytics) without re-introducing reorder hazards on a
+    // multi-threaded IO pool.
     override fun onFocus(firedOnSubscribe: Boolean) {
-        refreshNotificationState()
+        runOnSerialIOIfBackgroundThreading { refreshNotificationState() }
     }
 
     override fun onUnfocused() {
diff --git a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/internal/permissions/impl/NotificationPermissionController.kt b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/internal/permissions/impl/NotificationPermissionController.kt
index 0edc44f4e4..e219be9c13 100644
--- a/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/internal/permissions/impl/NotificationPermissionController.kt
+++ b/OneSignalSDK/onesignal/notifications/src/main/java/com/onesignal/notifications/internal/permissions/impl/NotificationPermissionController.kt
@@ -34,6 +34,7 @@ import com.onesignal.common.events.EventProducer
 import com.onesignal.common.threading.Waiter
 import com.onesignal.common.threading.WaiterWithValue
 import com.onesignal.common.threading.launchOnIO
+import com.onesignal.common.threading.runOnSerialIOIfBackgroundThreading
 import com.onesignal.core.internal.application.ApplicationLifecycleHandlerBase
 import com.onesignal.core.internal.application.IApplicationService
 import com.onesignal.core.internal.config.ConfigModelStore
@@ -84,16 +85,32 @@ internal class NotificationPermissionController(
     private fun registerPollingLifecycleListener() {
         _applicationService.addApplicationLifecycleHandler(
             object : ApplicationLifecycleHandlerBase() {
+                // The body reads `_configModelStore.model.foregroundFetchNotificationPermissionInterval`
+                // (a synchronized-map read against the live ConfigModel) and calls
+                // `pollingWaiter.wake()`, which dispatches a coroutine resume onto the IO pool
+                // via `channel.trySend` -> `ThreadPoolExecutor.execute`. Production OTel shows
+                // that on cold start the dispatcher / executor lazy chain itself stalls the
+                // main thread for many seconds when this is the first IO-pool consumer (see
+                // SDK-4507).
+                //
+                // Gated on SDK_BACKGROUND_THREADING via runOnSerialIOIfBackgroundThreading so we
+                // can A/B compare offloaded vs inline behavior. FF off retains the original
+                // semantics so existing tests that drive `onFocus` synchronously still observe
+                // the polling-interval update / wake immediately on the calling thread.
                 override fun onFocus(firedOnSubscribe: Boolean) {
                     super.onFocus(firedOnSubscribe)
-                    pollingWaitInterval = _configModelStore.model.foregroundFetchNotificationPermissionInterval
-                    pollingWaiter.wake()
+                    runOnSerialIOIfBackgroundThreading {
+                        pollingWaitInterval = _configModelStore.model.foregroundFetchNotificationPermissionInterval
+                        pollingWaiter.wake()
+                    }
                 }
 
                 override fun onUnfocused() {
                     super.onUnfocused()
-                    // Changing the polling interval to 1 day to effectively pause polling
-                    pollingWaitInterval = _configModelStore.model.backgroundFetchNotificationPermissionInterval
+                    runOnSerialIOIfBackgroundThreading {
+                        // Changing the polling interval to 1 day to effectively pause polling
+                        pollingWaitInterval = _configModelStore.model.backgroundFetchNotificationPermissionInterval
+                    }
                 }
             },
         )
diff --git a/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/internal/NotificationsManagerTests.kt b/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/internal/NotificationsManagerTests.kt
new file mode 100644
index 0000000000..cb93a3943a
--- /dev/null
+++ b/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/internal/NotificationsManagerTests.kt
@@ -0,0 +1,123 @@
+package com.onesignal.notifications.internal
+
+import androidx.test.core.app.ApplicationProvider
+import br.com.colman.kotest.android.extensions.robolectric.RobolectricTest
+import com.onesignal.common.threading.runOnSerialIOIfBackgroundThreading
+import com.onesignal.common.threading.suspendifyOnIO
+import com.onesignal.core.internal.application.IApplicationService
+import com.onesignal.debug.LogLevel
+import com.onesignal.debug.internal.logging.Logging
+import com.onesignal.notifications.internal.data.INotificationRepository
+import com.onesignal.notifications.internal.lifecycle.INotificationLifecycleService
+import com.onesignal.notifications.internal.permissions.INotificationPermissionController
+import com.onesignal.notifications.internal.restoration.INotificationRestoreWorkManager
+import com.onesignal.notifications.internal.summary.INotificationSummaryManager
+import com.onesignal.notifications.shadows.ShadowRoboNotificationManager
+import io.kotest.core.spec.style.FunSpec
+import io.mockk.every
+import io.mockk.just
+import io.mockk.mockk
+import io.mockk.mockkStatic
+import io.mockk.runs
+import io.mockk.unmockkStatic
+import io.mockk.verify
+import io.mockk.verifyOrder
+import org.robolectric.annotation.Config
+
+/**
+ * Regression coverage for the WorkManager-DB ANR observed in production
+ * (OTel sample insertId `9qy5s0ta0cwqwmb0`, vivo I2306 / Android 15: 30.5 s
+ * main-thread block at `NotificationRestoreWorkManager.beginEnqueueingWork`
+ * fired from `Activity.onStart`).
+ *
+ * Same Activity-lifecycle fan-out as SDK-4505: `onActivityStarted` -> `handleFocus` ->
+ * `applicationLifecycleNotifier.fire { onFocus(...) }` synchronously invokes every
+ * `IApplicationLifecycleHandler` on the main thread. `NotificationsManager.onFocus` was
+ * doing `WorkManager.enqueueUniqueWork` (which also lazily initializes the WorkManager
+ * SQLite store on first call) inline, and the SQLite write stalled the main thread on
+ * devices with slow / contended storage.
+ *
+ * The fix routes through `runOnSerialIOIfBackgroundThreading` — gated on the
+ * `SDK_BACKGROUND_THREADING` remote feature flag so we can A/B compare the offloaded
+ * behavior (FF on → serial IO dispatch) against the previous inline behavior (FF off →
+ * main thread) in production. These tests assert the dispatch contract on `onFocus`; the
+ * helper's two branches are tested in `:core` against `ThreadUtilsTests`, which has direct
+ * access to the internal `ThreadingMode` flag.
+ *
+ * `suspendifyOnIO` is also stubbed because `NotificationsManager`'s init block fires it for
+ * `deleteExpiredNotifications`; without the stub a real coroutine would leak past test
+ * teardown.
+ */
+@Config(
+    packageName = "com.onesignal.example",
+    shadows = [ShadowRoboNotificationManager::class],
+    sdk = [33],
+)
+@RobolectricTest
+class NotificationsManagerTests : FunSpec({
+
+    val threadUtilsPath = "com.onesignal.common.threading.ThreadUtilsKt"
+
+    beforeEach {
+        Logging.logLevel = LogLevel.NONE
+        ShadowRoboNotificationManager.reset()
+        mockkStatic(threadUtilsPath)
+        every { runOnSerialIOIfBackgroundThreading(any<() -> Unit>()) } just runs
+        every { suspendifyOnIO(any<suspend () -> Unit>()) } just runs
+    }
+
+    afterEach {
+        unmockkStatic(threadUtilsPath)
+    }
+
+    fun newManager(): NotificationsManager {
+        val mockAppService = mockk<IApplicationService>()
+        every { mockAppService.addApplicationLifecycleHandler(any()) } just runs
+        every { mockAppService.appContext } returns ApplicationProvider.getApplicationContext()
+
+        val permissionController = mockk<INotificationPermissionController>()
+        every { permissionController.subscribe(any()) } just runs
+
+        val restoreWorkManager = mockk<INotificationRestoreWorkManager>()
+        every { restoreWorkManager.beginEnqueueingWork(any(), any()) } just runs
+
+        val lifecycleService = mockk<INotificationLifecycleService>(relaxed = true)
+        val dataController = mockk<INotificationRepository>(relaxed = true)
+        val summaryManager = mockk<INotificationSummaryManager>(relaxed = true)
+
+        return NotificationsManager(
+            mockAppService,
+            permissionController,
+            restoreWorkManager,
+            lifecycleService,
+            dataController,
+            summaryManager,
+        )
+    }
+
+    test("onFocus dispatches refreshNotificationState through runOnSerialIOIfBackgroundThreading") {
+        val manager = newManager()
+
+        manager.onFocus(firedOnSubscribe = false)
+
+        verify(exactly = 1) { runOnSerialIOIfBackgroundThreading(any<() -> Unit>()) }
+    }
+
+    test("rapid onFocus burst dispatches each event through the gated helper in submission order") {
+        val manager = newManager()
+
+        // Two focus events in quick succession on the main thread (e.g. activity restart
+        // bouncing between activities). Both must route through the same gated helper in
+        // submission order — same defense the BackgroundManager burst test enforces for
+        // its `schedule`/`cancel` pair, ensuring future per-event work added here observes
+        // events in main-thread arrival order under the FF-on branch.
+        manager.onFocus(firedOnSubscribe = false)
+        manager.onFocus(firedOnSubscribe = false)
+
+        verify(exactly = 2) { runOnSerialIOIfBackgroundThreading(any<() -> Unit>()) }
+        verifyOrder {
+            runOnSerialIOIfBackgroundThreading(any<() -> Unit>())
+            runOnSerialIOIfBackgroundThreading(any<() -> Unit>())
+        }
+    }
+})
diff --git a/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/internal/permission/NotificationPermissionControllerTests.kt b/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/internal/permission/NotificationPermissionControllerTests.kt
index 59665fc216..63cd8442a8 100644
--- a/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/internal/permission/NotificationPermissionControllerTests.kt
+++ b/OneSignalSDK/onesignal/notifications/src/test/java/com/onesignal/notifications/internal/permission/NotificationPermissionControllerTests.kt
@@ -2,6 +2,8 @@ package com.onesignal.notifications.internal.permission
 
 import androidx.test.core.app.ApplicationProvider
 import br.com.colman.kotest.android.extensions.robolectric.RobolectricTest
+import com.onesignal.common.threading.OneSignalDispatchers
+import com.onesignal.common.threading.runOnSerialIOIfBackgroundThreading
 import com.onesignal.core.internal.application.IApplicationLifecycleHandler
 import com.onesignal.core.internal.application.IApplicationService
 import com.onesignal.core.internal.permissions.IRequestPermissionService
@@ -17,7 +19,13 @@ import io.kotest.matchers.shouldBe
 import io.mockk.every
 import io.mockk.just
 import io.mockk.mockk
+import io.mockk.mockkObject
+import io.mockk.mockkStatic
 import io.mockk.runs
+import io.mockk.unmockkObject
+import io.mockk.unmockkStatic
+import io.mockk.verify
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.delay
 import org.robolectric.annotation.Config
 
@@ -117,6 +125,93 @@ class NotificationPermissionControllerTests : FunSpec({
         handlerFired shouldBe false
     }
 
+    test("onFocus dispatches polling-interval update + waker through runOnSerialIOIfBackgroundThreading (SDK-4507)") {
+        // SDK-4507: the lifecycle-registered onFocus handler reads ConfigModel and calls
+        // Waiter.wake(), the latter of which dispatches a coroutine resume into the IO pool.
+        // On cold start this is the SDK's first OneSignalDispatchers consumer in the process,
+        // and the executor + dispatcher + scope lazy chain pinned the main thread for many
+        // seconds under sdk_background_threading. The fix routes through
+        // runOnSerialIOIfBackgroundThreading; verify that contract here.
+        //
+        // We stub the helper so the wrapped block does not run (we don't want a real
+        // pollingWaiter.wake() to spawn a real coroutine from this test). The FF branches of
+        // the helper itself are covered in :core's ThreadUtilsFeatureFlagTests, which has
+        // direct access to the internal ThreadingMode flag.
+        val threadUtilsPath = "com.onesignal.common.threading.ThreadUtilsKt"
+        mockkStatic(threadUtilsPath)
+        mockkObject(OneSignalDispatchers)
+        every { runOnSerialIOIfBackgroundThreading(any<() -> Unit>()) } just runs
+        every { OneSignalDispatchers.launchOnIO(any<suspend () -> Unit>()) } returns mockk<Job>(relaxed = true)
+
+        try {
+            val mockRequestPermissionService = mockk<IRequestPermissionService>()
+            every { mockRequestPermissionService.registerAsCallback(any(), any()) } just runs
+            val mockPreferenceService = mockk<IPreferencesService>()
+            val focusHandlerList = mutableListOf<IApplicationLifecycleHandler>()
+            val mockAppService = mockk<IApplicationService>()
+            every { mockAppService.addApplicationLifecycleHandler(any()) } answers {
+                focusHandlerList.add(firstArg<IApplicationLifecycleHandler>())
+            }
+            every { mockAppService.appContext } returns ApplicationProvider.getApplicationContext()
+
+            NotificationPermissionController(
+                mockAppService,
+                mockRequestPermissionService,
+                mockAppService,
+                mockPreferenceService,
+                MockHelper.configModelStore(),
+            )
+
+            for (focusHandler in focusHandlerList) {
+                focusHandler.onFocus(false)
+            }
+
+            // Only the polling lifecycle listener (registered inside the controller's init)
+            // routes through the gated helper, so we assert exactly 1 invocation here.
+            verify(exactly = 1) { runOnSerialIOIfBackgroundThreading(any<() -> Unit>()) }
+        } finally {
+            unmockkObject(OneSignalDispatchers)
+            unmockkStatic(threadUtilsPath)
+        }
+    }
+
+    test("onUnfocused dispatches polling-interval reset through runOnSerialIOIfBackgroundThreading (SDK-4507)") {
+        val threadUtilsPath = "com.onesignal.common.threading.ThreadUtilsKt"
+        mockkStatic(threadUtilsPath)
+        mockkObject(OneSignalDispatchers)
+        every { runOnSerialIOIfBackgroundThreading(any<() -> Unit>()) } just runs
+        every { OneSignalDispatchers.launchOnIO(any<suspend () -> Unit>()) } returns mockk<Job>(relaxed = true)
+
+        try {
+            val mockRequestPermissionService = mockk<IRequestPermissionService>()
+            every { mockRequestPermissionService.registerAsCallback(any(), any()) } just runs
+            val mockPreferenceService = mockk<IPreferencesService>()
+            val focusHandlerList = mutableListOf<IApplicationLifecycleHandler>()
+            val mockAppService = mockk<IApplicationService>()
+            every { mockAppService.addApplicationLifecycleHandler(any()) } answers {
+                focusHandlerList.add(firstArg<IApplicationLifecycleHandler>())
+            }
+            every { mockAppService.appContext } returns ApplicationProvider.getApplicationContext()
+
+            NotificationPermissionController(
+                mockAppService,
+                mockRequestPermissionService,
+                mockAppService,
+                mockPreferenceService,
+                MockHelper.configModelStore(),
+            )
+
+            for (focusHandler in focusHandlerList) {
+                focusHandler.onUnfocused()
+            }
+
+            verify(exactly = 1) { runOnSerialIOIfBackgroundThreading(any<() -> Unit>()) }
+        } finally {
+            unmockkObject(OneSignalDispatchers)
+            unmockkStatic(threadUtilsPath)
+        }
+    }
+
     test("NotificationPermissionController permission polling resumes when app gains focus") {
         // Given
         val mockRequestPermissionService = mockk<IRequestPermissionService>()
diff --git a/OneSignalSDK/onesignal/testhelpers/src/main/java/com/onesignal/mocks/IOMockHelper.kt b/OneSignalSDK/onesignal/testhelpers/src/main/java/com/onesignal/mocks/IOMockHelper.kt
index e3baea605e..adff0219df 100644
--- a/OneSignalSDK/onesignal/testhelpers/src/main/java/com/onesignal/mocks/IOMockHelper.kt
+++ b/OneSignalSDK/onesignal/testhelpers/src/main/java/com/onesignal/mocks/IOMockHelper.kt
@@ -1,8 +1,10 @@
 package com.onesignal.mocks
 
 import com.onesignal.common.threading.OneSignalDispatchers
+import com.onesignal.common.threading.runOnSerialIOIfBackgroundThreading
 import com.onesignal.common.threading.suspendifyOnIO
 import com.onesignal.common.threading.suspendifyOnMain
+import com.onesignal.common.threading.suspendifyOnSerialIO
 import io.kotest.core.listeners.AfterSpecListener
 import io.kotest.core.listeners.BeforeSpecListener
 import io.kotest.core.listeners.BeforeTestListener
@@ -26,13 +28,15 @@ import java.util.concurrent.atomic.AtomicInteger
  * Test helper that makes OneSignal's async threading behavior deterministic in unit tests.
  * Can be helpful to speed up unit tests by replacing all delay(x) or Thread.sleep(x).
  *
- * In production, `suspendifyOnIO`, `launchOnIO`, and `launchOnDefault` launch work on
- * background threads and return immediately. This causes tests to require arbitrary delays
- * (e.g., delay(50)) to wait for async work to finish.
+ * In production, `suspendifyOnIO`, `suspendifyOnSerialIO`, `launchOnIO`, `launchOnSerialIO`,
+ * and `launchOnDefault` launch work on background threads and return immediately. This causes
+ * tests to require arbitrary delays (e.g., delay(50)) to wait for async work to finish.
  *
  * This helper avoids that by:
- *  - Mocking `suspendifyOnIO`, `suspendifyOnMain`, and `OneSignalDispatchers.launchOnIO` /
- *    `launchOnDefault` so their blocks run immediately
+ *  - Mocking `suspendifyOnIO`, `suspendifyOnSerialIO`, `suspendifyOnMain`,
+ *    `runOnSerialIOIfBackgroundThreading`, and
+ *    `OneSignalDispatchers.launchOnIO` / `launchOnSerialIO` / `launchOnDefault` so their
+ *    blocks run immediately
  *  - Completing a `CompletableDeferred` when the async block finishes
  *  - Providing `awaitIO()` so tests can explicitly wait for all async work without sleeps
  *
@@ -126,6 +130,20 @@ object IOMockHelper : BeforeSpecListener, AfterSpecListener, BeforeTestListener,
             trackAsyncWork(block)
         }
 
+        every { suspendifyOnSerialIO(any<suspend () -> Unit>()) } answers {
+            val block = firstArg<suspend () -> Unit>()
+            trackAsyncWork(block)
+        }
+
+        // Run the block inline on the test thread so callers see the same observable behavior
+        // as the FF-off branch in production. Tests that need to exercise the FF-on branch can
+        // override this with their own `every { runOnSerialIOIfBackgroundThreading(...) }`
+        // (e.g. routing through `suspendifyOnSerialIO` + `trackAsyncWork`).
+        every { runOnSerialIOIfBackgroundThreading(any<() -> Unit>()) } answers {
+            val block = firstArg<() -> Unit>()
+            block()
+        }
+
         every { suspendifyOnMain(any<suspend () -> Unit>()) } answers {
             val block = firstArg<suspend () -> Unit>()
             trackAsyncWork(block)
@@ -138,6 +156,13 @@ object IOMockHelper : BeforeSpecListener, AfterSpecListener, BeforeTestListener,
             mockk<Job>(relaxed = true)
         }
 
+        every { OneSignalDispatchers.launchOnSerialIO(any<suspend () -> Unit>()) } answers {
+            val block = firstArg<suspend () -> Unit>()
+            trackAsyncWork(block)
+            // Return a mock Job (launchOnSerialIO returns a Job)
+            mockk<Job>(relaxed = true)
+        }
+
         every { OneSignalDispatchers.launchOnDefault(any<suspend () -> Unit>()) } answers {
             val block = firstArg<suspend () -> Unit>()
             trackAsyncWork(block)