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..11aaf3ae0e 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,21 @@ object OneSignalDispatchers {
         }
     }
 
+    /** Single-thread executor for order-sensitive lifecycle work (focus / unfocus handlers). */
+    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 +136,17 @@ object OneSignalDispatchers {
         }
     }
 
+    val SerialIO: CoroutineDispatcher by lazy {
+        try {
+            serialIOExecutor.asCoroutineDispatcher()
+        } catch (e: Exception) {
+            // Fall back to a limitedParallelism(1) view of Dispatchers.IO so submissions stay serialized.
+            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 +155,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 +167,26 @@ object OneSignalDispatchers {
         return DefaultScope.launch { block() }
     }
 
+    /** Launches [block] on the single-thread serial IO dispatcher (FIFO across all callers). */
+    fun launchOnSerialIO(block: suspend () -> Unit): Job {
+        return SerialIOScope.launch { block() }
+    }
+
     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 +194,39 @@ 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()
     }
+
+    // internal so tests can exercise the failure branch (when `isShutdown()` itself throws,
+    // which happens when the lazy initializer threw and re-throws on every access).
+    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..f6d37aca9a 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,38 @@ fun suspendifyOnDefault(block: suspend () -> Unit) {
     suspendifyWithCompletion(useIO = false, block = block, onComplete = null)
 }
 
+/**
+ * Runs [block] on the single-thread serial IO dispatcher. Tasks from any thread execute
+ * one-at-a-time in submission order — the entry point for lifecycle handlers that need to
+ * preserve event ordering. Always routes off-main regardless of [ThreadingMode.useBackgroundThreading].
+ *
+ * Capture time-sensitive state (timestamps, "current" snapshots) on the caller's thread
+ * before invoking — the block itself may run later under load.
+ */
+fun suspendifyOnSerialIO(block: suspend () -> Unit) {
+    OneSignalDispatchers.launchOnSerialIO {
+        try {
+            block()
+        } catch (e: Exception) {
+            Logging.error("Exception in suspendifyOnSerialIO", e)
+        }
+    }
+}
+
+/**
+ * FF-gated rollout helper for lifecycle offload work. When [ThreadingMode.useBackgroundThreading]
+ * is on, dispatches [block] to the serial IO thread; when off, runs it inline so the control
+ * cohort retains the original behavior. The block is non-suspending so the FF-off branch doesn't
+ * need a [kotlinx.coroutines.runBlocking].
+ */
+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..eace962553 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,7 @@ import android.content.ComponentName
 import android.content.Context
 import android.content.pm.PackageManager
 import androidx.core.content.ContextCompat
+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.IBackgroundManager
@@ -70,12 +71,15 @@ internal class BackgroundManager(
         _applicationService.addApplicationLifecycleHandler(this)
     }
 
+    // JobScheduler.cancel/schedule are synchronous Binder calls; on some devices they block the
+    // main thread for seconds (SDK-4505). Offload to SerialIO so a rapid unfocus -> focus burst
+    // still runs cancel-then-schedule in submission order. FF gates the rollout.
     override fun onFocus(firedOnSubscribe: Boolean) {
-        cancelSyncTask()
+        runOnSerialIOIfBackgroundThreading { cancelSyncTask() }
     }
 
     override fun onUnfocused() {
-        scheduleBackground()
+        runOnSerialIOIfBackgroundThreading { 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..83359b48e8 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,22 @@ internal class FeatureFlagsRefreshService(
         // Foreground-at-subscribe is handled by [IApplicationService.addApplicationLifecycleHandler] (fires onFocus).
     }
 
+    // restartForegroundPolling calls launchOnIO; on cold start that can be the first
+    // OneSignalDispatchers consumer and pay the lazy-chain init cost on the main thread
+    // (SDK-4506). SerialIO also preserves order with the matching onUnfocused cancel.
     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 service instance (same monitor
+            // restartForegroundPolling acquires), not on the no-receiver lambda.
+            synchronized(this@FeatureFlagsRefreshService) {
+                pollJob?.cancel()
+                pollJob = null
+                pollingAppId = null
+            }
         }
     }
 
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..43ff933811 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
@@ -103,6 +104,18 @@ internal class SessionService(
      * the `onSessionStarted()` callback here, so fire it when they themselves subscribe.
      */
     override fun onFocus(firedOnSubscribe: Boolean) {
+        // Capture focus time on the caller's thread so session timestamps reflect lifecycle
+        // arrival, not dispatcher latency (SDK-4506).
+        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 +134,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 +142,27 @@ 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() {
+        // Capture on the caller's thread so activeDuration is unaffected by dispatcher latency.
+        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..66add67b1e 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") {
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..a14d48c489 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,11 @@ internal class NotificationsManager(
         }
     }
 
+    // refreshNotificationState drives WorkManager DB I/O; the first call lazily inits WorkManager
+    // (opens/migrates its SQLite store) and can block the main thread for seconds on slow storage
+    // (SDK-4506).
     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..819da63f24 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,23 @@ internal class NotificationPermissionController(
     private fun registerPollingLifecycleListener() {
         _applicationService.addApplicationLifecycleHandler(
             object : ApplicationLifecycleHandlerBase() {
+                // pollingWaiter.wake() dispatches onto the IO pool; on cold start it can be the
+                // first OneSignalDispatchers consumer and pay the lazy-chain init cost on the
+                // main thread (SDK-4506).
                 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)