diff --git a/.github/workflows/add-to-project.yml b/.github/workflows/add-to-project.yml
deleted file mode 100644
index 0607710..0000000
--- a/.github/workflows/add-to-project.yml
+++ /dev/null
@@ -1,14 +0,0 @@
-name: Add PR to Mobile SDK Board
-
-on:
-  pull_request:
-    types: [opened]
-
-jobs:
-  add-to-project:
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/add-to-project@v1
-        with:
-          project-url: https://github.com/orgs/metarouterio/projects/1
-          github-token: ${{ secrets.ADD_TO_PROJECT_PAT }}
diff --git a/metarouter-sdk/src/main/java/com/metarouter/analytics/MetaRouterAnalyticsClient.kt b/metarouter-sdk/src/main/java/com/metarouter/analytics/MetaRouterAnalyticsClient.kt
index 18fb3c1..c04cfe0 100644
--- a/metarouter-sdk/src/main/java/com/metarouter/analytics/MetaRouterAnalyticsClient.kt
+++ b/metarouter-sdk/src/main/java/com/metarouter/analytics/MetaRouterAnalyticsClient.kt
@@ -152,7 +152,8 @@ class MetaRouterAnalyticsClient private constructor(
                 val diskStore = injectedDiskStore ?: EventDiskStore.create(context)
                 val pQueue = PersistableEventQueue(
                     maxCapacity = options.maxQueueEvents,
-                    diskStore = diskStore
+                    diskStore = diskStore,
+                    maxOfflineDiskEvents = options.maxOfflineDiskEvents
                 )
                 pQueue.rehydrate()
                 eventQueue = pQueue
@@ -182,6 +183,7 @@ class MetaRouterAnalyticsClient private constructor(
                     circuitBreaker.reset()
                     dispatcher.resumeFromOffline()
                     scope.launch { dispatcher.flush() }
+                    scope.launch { persistableEventQueue?.drainDiskOverflowToNetwork(dispatcher) }
                 } else {
                     dispatcher.pauseForOffline()
                 }
@@ -192,8 +194,23 @@ class MetaRouterAnalyticsClient private constructor(
                 dispatcher.pauseForOffline()
             }
 
+            // If online at launch and unsent events exist on disk from a previous session,
+            // drain them directly to the network.
+            if (networkMonitor.isConnected && persistableEventQueue?.hasOverflowData() == true) {
+                scope.launch { persistableEventQueue?.drainDiskOverflowToNetwork(dispatcher) }
+            }
+
             startEventProcessor()
 
+            // After each successful flush, drain any overflow disk events
+            persistableEventQueue?.let { pQueue ->
+                dispatcher.onFlushComplete = {
+                    if (networkMonitor.isConnected && pQueue.hasOverflowData()) {
+                        scope.launch { pQueue.drainDiskOverflowToNetwork(dispatcher) }
+                    }
+                }
+            }
+
             dispatcher.start()
 
             // Register best-effort terminate flush via ComponentCallbacks2
diff --git a/metarouter-sdk/src/main/java/com/metarouter/analytics/dispatcher/Dispatcher.kt b/metarouter-sdk/src/main/java/com/metarouter/analytics/dispatcher/Dispatcher.kt
index 772d958..783b72f 100644
--- a/metarouter-sdk/src/main/java/com/metarouter/analytics/dispatcher/Dispatcher.kt
+++ b/metarouter-sdk/src/main/java/com/metarouter/analytics/dispatcher/Dispatcher.kt
@@ -60,6 +60,12 @@ class Dispatcher(
      */
     var onFatalConfigError: ((Int) -> Unit)? = null
 
+    /**
+     * Callback invoked after a successful flush cycle completes.
+     * Used to trigger overflow disk drain when events spill to disk while online.
+     */
+    var onFlushComplete: (() -> Unit)? = null
+
     // ===== Lifecycle =====
 
     /**
@@ -169,6 +175,9 @@ class Dispatcher(
 
         try {
             processUntilEmpty()
+            if (!networkPaused) {
+                onFlushComplete?.invoke()
+            }
         } finally {
             isFlushing.set(false)
         }
@@ -191,6 +200,56 @@ class Dispatcher(
         )
     }
 
+    // ===== Direct Send (for disk overflow drain) =====
+
+    /**
+     * Send a batch of events directly to the network, bypassing the memory queue.
+     * Used by PersistableEventQueue.drainDiskOverflowToNetwork() to flush overflow
+     * events from disk without loading them into the memory queue.
+     *
+     * Note: intentionally does not update the circuit breaker or retry counters.
+     * The overflow drain is an independent pipeline — it stops on first failure and
+     * retries on the next online transition or flush cycle, not via the dispatcher's
+     * retry/backoff machinery.
+     *
+     * @return NetworkResponse on HTTP completion, null on encoding or network error
+     */
+    suspend fun sendBatchDirect(events: List<EnrichedEventPayload>): NetworkResponse? {
+        if (events.isEmpty()) return NetworkResponse(200, emptyMap(), null)
+
+        val sentAt = getIso8601Timestamp()
+        val batch = events.map { it.copy(sentAt = sentAt) }
+
+        val url = "${options.getNormalizedIngestionHost()}${config.endpointPath}"
+        val body: ByteArray
+        try {
+            val payload = BatchPayload(batch = batch)
+            body = json.encodeToString(payload).toByteArray(Charsets.UTF_8)
+        } catch (e: Exception) {
+            Logger.error("Failed to encode direct batch of ${batch.size} events: ${e.message}")
+            return null
+        }
+
+        val headers = mutableMapOf<String, String>()
+        if (tracingEnabled) {
+            headers["Trace"] = "true"
+        }
+
+        Logger.log("Making API call to: $url (overflow drain, ${batch.size} events)")
+
+        return try {
+            networkClient.postJson(
+                url = url,
+                body = body,
+                timeoutMs = config.timeoutMs,
+                additionalHeaders = if (headers.isNotEmpty()) headers else null
+            )
+        } catch (e: Exception) {
+            Logger.warn("Direct batch send failed: ${e.message}")
+            null
+        }
+    }
+
     // ===== Internal =====
 
     /**
@@ -209,7 +268,11 @@ class Dispatcher(
 
     private suspend fun processUntilEmpty() {
         if (networkPaused) {
-            Logger.log("Dispatcher skipping flush — device is offline")
+            if (queue.flushToOfflineStorage()) {
+                Logger.log("Flushed queue to offline storage")
+            } else {
+                Logger.log("Dispatcher skipping flush — device is offline")
+            }
             return
         }
 
@@ -282,8 +345,8 @@ class Dispatcher(
         response: NetworkResponse,
         batch: List<EnrichedEventPayload>
     ): FlushAction {
-        return when (response.statusCode) {
-            in 200..299 -> {
+        return when (ResponseCategory.from(response.statusCode)) {
+            ResponseCategory.SUCCESS -> {
                 consecutiveRetries.set(0)
                 circuitBreaker.onSuccess()
                 val current = maxBatchSize.get()
@@ -294,7 +357,7 @@ class Dispatcher(
                 Logger.log("API call successful")
                 FlushAction.Continue
             }
-            in 500..599, 408 -> {
+            ResponseCategory.SERVER_ERROR -> {
                 consecutiveRetries.incrementAndGet()
                 circuitBreaker.onFailure()
                 queue.requeueToFront(batch)
@@ -303,7 +366,7 @@ class Dispatcher(
                 Logger.warn("Server error ${response.statusCode}, will retry ${batch.size} event(s) in ${waitMs}ms (circuit: ${circuitBreaker.getState()}, retry #${consecutiveRetries.get()})")
                 FlushAction.RetryAfter(waitMs)
             }
-            429 -> {
+            ResponseCategory.RATE_LIMITED -> {
                 consecutiveRetries.incrementAndGet()
                 circuitBreaker.onFailure()
                 queue.requeueToFront(batch)
@@ -313,7 +376,7 @@ class Dispatcher(
                 Logger.warn("Rate limited (429), will retry ${batch.size} event(s) in ${waitMs}ms (circuit: ${circuitBreaker.getState()}, retry #${consecutiveRetries.get()})")
                 FlushAction.RetryAfter(waitMs)
             }
-            413 -> {
+            ResponseCategory.PAYLOAD_TOO_LARGE -> {
                 circuitBreaker.onNonRetryable()
                 val currentSize = maxBatchSize.get()
                 if (currentSize > 1) {
@@ -327,23 +390,18 @@ class Dispatcher(
                     FlushAction.Stop
                 }
             }
-            401, 403, 404 -> {
+            ResponseCategory.FATAL_CONFIG -> {
                 Logger.error("Fatal configuration error: ${response.statusCode} - clearing queue and stopping")
                 queue.clear()
                 stop()
                 onFatalConfigError?.invoke(response.statusCode)
                 FlushAction.Stop
             }
-            in 400..499 -> {
+            ResponseCategory.CLIENT_ERROR -> {
                 circuitBreaker.onNonRetryable()
                 Logger.warn("Client error ${response.statusCode} - dropping batch of ${batch.size} events")
                 FlushAction.Continue
             }
-            else -> {
-                circuitBreaker.onNonRetryable()
-                Logger.warn("Unexpected status ${response.statusCode} - continuing")
-                FlushAction.Continue
-            }
         }
     }
 
@@ -368,7 +426,7 @@ class Dispatcher(
 }
 
 @kotlinx.serialization.Serializable
-private data class BatchPayload(
+internal data class BatchPayload(
     val batch: List<EnrichedEventPayload>
 )
 
@@ -378,6 +436,31 @@ private sealed class FlushAction {
     data object Stop : FlushAction()
 }
 
+/**
+ * Classifies an HTTP status code into a response category.
+ * Shared by the main dispatcher flush loop and the overflow drain.
+ */
+internal enum class ResponseCategory {
+    SUCCESS,
+    SERVER_ERROR,
+    RATE_LIMITED,
+    PAYLOAD_TOO_LARGE,
+    FATAL_CONFIG,
+    CLIENT_ERROR;
+
+    companion object {
+        fun from(statusCode: Int): ResponseCategory = when (statusCode) {
+            in 200..299 -> SUCCESS
+            in 500..599, 408 -> SERVER_ERROR
+            429 -> RATE_LIMITED
+            413 -> PAYLOAD_TOO_LARGE
+            401, 403, 404 -> FATAL_CONFIG
+            in 400..499 -> CLIENT_ERROR
+            else -> CLIENT_ERROR // Unknown status codes are non-retryable (drop, don't retry)
+        }
+    }
+}
+
 data class DispatcherDebugInfo(
     val isRunning: Boolean,
     val maxBatchSize: Int,
diff --git a/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/EventQueueInterface.kt b/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/EventQueueInterface.kt
index 454cd60..01597cf 100644
--- a/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/EventQueueInterface.kt
+++ b/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/EventQueueInterface.kt
@@ -12,4 +12,13 @@ interface EventQueueInterface {
     fun drain(max: Int): List<EnrichedEventPayload>
     fun requeueToFront(events: List<EnrichedEventPayload>)
     fun clear()
+
+    /**
+     * Flush all events to offline storage. Called by the dispatcher when
+     * a flush is triggered but the device is offline.
+     * Default implementation is a no-op (returns false).
+     *
+     * @return true if events were flushed to offline storage
+     */
+    fun flushToOfflineStorage(): Boolean = false
 }
diff --git a/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/PersistableEventQueue.kt b/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/PersistableEventQueue.kt
index d43382d..a81d09f 100644
--- a/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/PersistableEventQueue.kt
+++ b/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/PersistableEventQueue.kt
@@ -1,5 +1,7 @@
 package com.metarouter.analytics.queue
 
+import com.metarouter.analytics.dispatcher.Dispatcher
+import com.metarouter.analytics.dispatcher.ResponseCategory
 import com.metarouter.analytics.storage.EventDiskStore
 import com.metarouter.analytics.storage.QueueSnapshot
 import com.metarouter.analytics.types.EnrichedEventPayload
@@ -12,17 +14,23 @@ import java.util.TimeZone
 import java.util.concurrent.atomic.AtomicBoolean
 
 /**
- * Event queue with disk persistence as a safety net.
- * Uses composition: wraps an internal memory queue and adds byte tracking + disk I/O.
+ * Event queue with a single disk store for persistence and offline overflow.
+ * Memory queue is the hot buffer; disk is the overflow/crash-safety backing.
  *
- * Write-behind model:
- * - enqueue() and drain() are memory-only (no disk I/O)
- * - flushToDisk() writes a full snapshot of current memory state
- * - rehydrate() loads events from disk once per process lifetime
+ * Model:
+ * - enqueue() and drain() are memory-only (no disk I/O) in the common case.
+ * - When memory queue hits capacity (event count OR byte size), the entire queue
+ *   flushes to disk and memory resets to empty. Same triggers swap destination
+ *   when the device is offline (see [flushToOfflineStorage]).
+ * - flushToDisk() appends current memory queue to disk (crash-safety snapshot).
+ * - drainDiskOverflowToNetwork() sends disk events directly to the network via
+ *   Dispatcher.sendBatchDirect, bypassing the memory queue.
  *
- * Capacity model:
- * - Single shared cap: [maxCapacity] events OR [maxCapacityBytes] total, whichever is hit first
- * - Flush threshold: [flushThresholdEvents] events OR [flushThresholdBytes], whichever is hit first
+ * Concurrency:
+ * - All memory-queue operations are guarded by `synchronized(this)` via @Synchronized.
+ * - Disk operations happen under a separate [diskLock] so the suspend drain can
+ *   release it during network calls without blocking enqueue/drain of the memory queue.
+ * - Lock order is always `this` -> `diskLock` (never the reverse).
  */
 class PersistableEventQueue(
     private val maxCapacity: Int = 2000,
@@ -30,7 +38,8 @@ class PersistableEventQueue(
     private val flushThresholdEvents: Int = 500,
     private val flushThresholdBytes: Long = 2L * 1024 * 1024,
     private val maxCapacityBytes: Long = 5L * 1024 * 1024,
-    private val eventTTLMs: Long = 7L * 24 * 60 * 60 * 1000
+    private val eventTTLMs: Long = 7L * 24 * 60 * 60 * 1000,
+    private val maxOfflineDiskEvents: Int = 10000
 ) : EventQueueInterface {
 
     companion object {
@@ -42,6 +51,9 @@ class PersistableEventQueue(
             }
         }
 
+        /** Write a checkpoint every N successful drain batches for crash safety. */
+        private const val DRAIN_CHECKPOINT_BATCHES = 10
+
         /**
          * Reset the rehydration flag so a subsequent initialize() will rehydrate from disk.
          * Called during SDK reset and in tests.
@@ -60,25 +72,39 @@ class PersistableEventQueue(
     // Cache of per-event byte sizes (parallel to memoryQueue order)
     private val eventSizes = ArrayDeque<Long>()
 
+    // Guards against concurrent drainDiskOverflowToNetwork invocations
+    private val isDraining = AtomicBoolean(false)
+
+    // True when unsent events exist on disk. Cheap volatile flag used to gate drain attempts.
+    // Initialized from file existence (not a full file parse).
+    @Volatile
+    private var hasOverflowData = try { diskStore.exists() } catch (_: Exception) { false }
+
+    // Lock for disk store access. The suspend drain releases this lock during network calls,
+    // so it can't hold the class monitor. Memory queue methods acquire `this` first and
+    // `diskLock` second (when they touch disk). Drain acquires only `diskLock`. No deadlock.
+    private val diskLock = Any()
+
     @Synchronized
     override fun size(): Int = memoryQueue.size
 
     /**
-     * Enqueue an event to memory. No disk I/O.
-     * Drops oldest events if event count or byte capacity is exceeded.
+     * Enqueue an event to memory.
+     * When at capacity (event count or byte size), flushes the entire memory queue
+     * to disk. Events are only dropped if capacity is hit and disk operations fail.
      */
     @Synchronized
     override fun enqueue(event: EnrichedEventPayload) {
         val eventSize = estimateEventSize(event)
 
-        // Drop oldest while over byte capacity (before adding)
-        while (estimatedBytes + eventSize > maxCapacityBytes && memoryQueue.isNotEmpty()) {
-            dropOldest("byte capacity")
+        // Over byte capacity: flush memory queue to disk
+        if (estimatedBytes + eventSize > maxCapacityBytes && memoryQueue.isNotEmpty()) {
+            flushMemoryToDiskInternal()
         }
 
-        // Drop oldest if at event count capacity
+        // At event count capacity: flush memory queue to disk
         if (memoryQueue.size >= maxCapacity) {
-            dropOldest("event count capacity")
+            flushMemoryToDiskInternal()
         }
 
         memoryQueue.addLast(event)
@@ -101,29 +127,31 @@ class PersistableEventQueue(
     }
 
     /**
-     * Requeue events to front. Updates byte size tracking.
+     * Requeue events to front (for retry after a failed send).
+     * When at capacity, flushes existing memory queue to disk to make room. The requeued
+     * (older) events take priority at the front since they were drained first and a send
+     * already started for them. If the requeue batch itself exceeds capacity mid-loop,
+     * we flush again so no requeued events get dropped.
      */
     @Synchronized
     override fun requeueToFront(events: List<EnrichedEventPayload>) {
+        // Initial flush: if adding these events would exceed capacity, flush current memory first.
+        if ((memoryQueue.size + events.size > maxCapacity ||
+             estimatedBytes + events.sumOf { estimateEventSize(it) } > maxCapacityBytes) &&
+            memoryQueue.isNotEmpty()) {
+            flushMemoryToDiskInternal()
+        }
+
         events.asReversed().forEach { event ->
             val eventSize = estimateEventSize(event)
 
-            // Drop newest while over byte capacity
-            while (estimatedBytes + eventSize > maxCapacityBytes && memoryQueue.isNotEmpty()) {
-                val removedSize = eventSizes.removeLastOrNull() ?: 0L
-                memoryQueue.removeLastOrNull()?.let {
-                    estimatedBytes -= removedSize
-                    Logger.warn("Queue byte capacity reached during requeue - dropped newest event (messageId: ${it.messageId})")
-                }
+            // Mid-loop capacity: the requeue batch itself exceeds capacity after earlier iterations
+            // filled the queue. Flush to disk instead of dropping requeued events.
+            if (estimatedBytes + eventSize > maxCapacityBytes && memoryQueue.isNotEmpty()) {
+                flushMemoryToDiskInternal()
             }
-
-            // Drop newest if at event count capacity
             if (memoryQueue.size >= maxCapacity) {
-                val removedSize = eventSizes.removeLastOrNull() ?: 0L
-                memoryQueue.removeLastOrNull()?.let {
-                    estimatedBytes -= removedSize
-                    Logger.warn("Queue event capacity reached during requeue - dropped newest event (messageId: ${it.messageId})")
-                }
+                flushMemoryToDiskInternal()
             }
 
             memoryQueue.addFirst(event)
@@ -133,7 +161,7 @@ class PersistableEventQueue(
     }
 
     /**
-     * Clear all events from memory and delete the disk snapshot.
+     * Clear all events from memory and delete the disk file.
      */
     @Synchronized
     override fun clear() {
@@ -141,31 +169,40 @@ class PersistableEventQueue(
         memoryQueue.clear()
         eventSizes.clear()
         estimatedBytes = 0L
-        diskStore.delete()
+        synchronized(diskLock) {
+            deleteDiskStore()
+        }
         if (count > 0) {
             Logger.log("Cleared $count events from queue and deleted disk snapshot")
         }
     }
 
     /**
-     * Flush current memory state to disk as a full overwrite snapshot.
-     * Skips the write and deletes any existing snapshot if the queue is empty.
+     * Flush memory queue to disk when the dispatcher triggers a flush while offline.
+     * Returns true if events were flushed to disk.
+     */
+    @Synchronized
+    override fun flushToOfflineStorage(): Boolean {
+        if (memoryQueue.isEmpty()) return false
+        flushMemoryToDiskInternal()
+        return true
+    }
+
+    /**
+     * Best-effort crash-safety flush: appends current memory queue to disk.
+     * Called on app background / onTrimMemory.
      */
     @Synchronized
     fun flushToDisk() {
-        if (memoryQueue.isEmpty()) {
-            diskStore.delete()
-            return
-        }
-        val events = memoryQueue.toList()
-        val snapshot = QueueSnapshot(version = 1, events = events)
-        diskStore.write(snapshot)
+        if (memoryQueue.isEmpty()) return
+        flushMemoryToDiskInternal()
     }
 
     /**
-     * Rehydrate events from disk into memory. Only executes once per process.
-     * Loaded events count toward capacity. Excess events are trimmed (keep newest).
-     * Deletes the disk file after successful load.
+     * Called once per process on startup. With the consolidated disk store, events
+     * persisted across sessions stay on disk and drain to the network via
+     * [drainDiskOverflowToNetwork]. Memory queue starts empty — no loading needed.
+     * [hasOverflowData] reflects whether disk has unsent events (set in constructor).
      */
     @Synchronized
     fun rehydrate() {
@@ -173,81 +210,197 @@ class PersistableEventQueue(
             Logger.log("Rehydration already completed this process - skipping")
             return
         }
+        if (hasOverflowData) {
+            Logger.log("Unsent events present on disk — will drain on next online transition")
+        }
+    }
 
-        val snapshot = diskStore.read() ?: return
+    /**
+     * Check if the flush threshold has been reached (either event count or byte size).
+     */
+    @Synchronized
+    fun shouldFlushToDisk(): Boolean {
+        return memoryQueue.size >= flushThresholdEvents || estimatedBytes >= flushThresholdBytes
+    }
 
-        var events = snapshot.events
-        if (events.isEmpty()) {
-            diskStore.delete()
-            return
-        }
+    /**
+     * Get the estimated total byte size of events in memory.
+     */
+    @Synchronized
+    fun estimatedSizeBytes(): Long = estimatedBytes
+
+    /**
+     * Check if there is unsent data on disk worth draining.
+     * Cheap check — reads a volatile flag, no disk I/O.
+     */
+    fun hasOverflowData(): Boolean = hasOverflowData
+
+    // ===== Offline Overflow =====
 
-        // Drop events older than TTL (keep newest)
-        if (eventTTLMs > 0) {
+    /**
+     * Drain disk events directly to the network in batches via [Dispatcher.sendBatchDirect].
+     * Does NOT load events into the memory queue. Called on offline->online transitions
+     * and after successful online flushes via onFlushComplete.
+     *
+     * Reads the entire disk file once, deletes it, then iterates batches in-memory.
+     * Writes a checkpoint every [DRAIN_CHECKPOINT_BATCHES] successful batches for crash safety.
+     * On failure, writes remaining events back to disk.
+     *
+     * @return number of events successfully drained
+     */
+    suspend fun drainDiskOverflowToNetwork(dispatcher: Dispatcher): Int {
+        // Bail early if offline — no point burning network call timeouts.
+        if (dispatcher.isNetworkPaused()) return 0
+        if (!isDraining.compareAndSet(false, true)) return 0
+
+        var totalDrained = 0
+        var currentBatchSize = 100
+
+        try {
+            // Read entire file once and delete — O(n) instead of O(n²)
+            val allEvents = synchronized(diskLock) {
+                val snapshot = diskStore.read() ?: return 0
+                deleteDiskStore()
+                snapshot.events
+            }
+
+            // Filter expired events
             val now = System.currentTimeMillis()
-            val beforeTTL = events.size
-            events = events.filter { !isExpired(it, now) }
-            val dropped = beforeTTL - events.size
-            if (dropped > 0) {
-                Logger.warn("Rehydration: dropped $dropped event(s) older than ${eventTTLMs / (24 * 60 * 60 * 1000)}d TTL")
+            val events = if (eventTTLMs > 0) {
+                allEvents.filter { !isExpired(it, now) }
+            } else {
+                allEvents
             }
-        }
 
-        // Trim to event count capacity (keep newest)
-        if (events.size > maxCapacity) {
-            Logger.warn("Disk snapshot has ${events.size} events, trimming to $maxCapacity (keeping newest)")
-            events = events.drop(events.size - maxCapacity)
-        }
+            if (events.isEmpty()) return 0
 
-        // Trim by byte capacity (keep newest)
-        var totalBytes = 0L
-        val sizedEvents = events.map { it to estimateEventSize(it) }
-        val fittingEvents = mutableListOf<Pair<EnrichedEventPayload, Long>>()
+            // Track position via index — avoids O(n) removeAt(0) on each batch
+            var offset = 0
+            var batchesSinceCheckpoint = 0
+            fun remaining() = events.subList(offset, events.size)
 
-        for ((event, size) in sizedEvents.asReversed()) {
-            if (totalBytes + size > maxCapacityBytes) break
-            totalBytes += size
-            fittingEvents.add(event to size)
-        }
-        fittingEvents.reverse()
+            while (offset < events.size) {
+                val batchEnd = minOf(offset + currentBatchSize, events.size)
+                val batch = events.subList(offset, batchEnd)
 
-        if (fittingEvents.size < events.size) {
-            Logger.warn("Disk snapshot exceeds byte capacity, trimmed from ${events.size} to ${fittingEvents.size} events")
-        }
+                val response = dispatcher.sendBatchDirect(batch)
 
-        // Load into memory
-        for ((event, size) in fittingEvents) {
-            memoryQueue.addLast(event)
-            eventSizes.addLast(size)
-            estimatedBytes += size
-        }
+                if (response == null) {
+                    writeCheckpoint(remaining())
+                    Logger.warn("Offline overflow drain paused — network error ($totalDrained drained so far)")
+                    return totalDrained
+                }
 
-        Logger.log("Rehydrated ${fittingEvents.size} events from disk")
-        diskStore.delete()
+                when (ResponseCategory.from(response.statusCode)) {
+                    ResponseCategory.SUCCESS -> {
+                        offset += batch.size
+                        totalDrained += batch.size
+                        batchesSinceCheckpoint++
+                        if (currentBatchSize < 100) {
+                            currentBatchSize = minOf(currentBatchSize * 2, 100)
+                        }
+                        // Periodic checkpoint for crash safety
+                        if (batchesSinceCheckpoint >= DRAIN_CHECKPOINT_BATCHES && offset < events.size) {
+                            writeCheckpoint(remaining())
+                            batchesSinceCheckpoint = 0
+                        }
+                    }
+                    ResponseCategory.PAYLOAD_TOO_LARGE -> {
+                        if (currentBatchSize > 1) {
+                            currentBatchSize = maxOf(1, currentBatchSize / 2)
+                            Logger.warn("Overflow drain: payload too large (413) — reduced batch to $currentBatchSize")
+                        } else {
+                            Logger.warn("Overflow drain: dropping oversized event at batchSize=1 (messageId: ${batch.first().messageId})")
+                            offset++
+                        }
+                    }
+                    ResponseCategory.FATAL_CONFIG -> {
+                        Logger.error("Overflow drain: fatal config error ${response.statusCode} — discarding ${events.size - offset} overflow events")
+                        // Notify the dispatcher so the main pipeline also shuts down. Matches
+                        // the main flush path's handling of auth/config failures.
+                        dispatcher.onFatalConfigError?.invoke(response.statusCode)
+                        return totalDrained
+                    }
+                    ResponseCategory.CLIENT_ERROR -> {
+                        Logger.warn("Overflow drain: client error ${response.statusCode} — dropping batch of ${batch.size}")
+                        offset += batch.size
+                    }
+                    ResponseCategory.SERVER_ERROR, ResponseCategory.RATE_LIMITED -> {
+                        writeCheckpoint(remaining())
+                        Logger.warn("Offline overflow drain paused — ${response.statusCode} ($totalDrained drained so far)")
+                        return totalDrained
+                    }
+                }
+            }
+
+            // All events drained successfully
+            synchronized(diskLock) { deleteDiskStore() }
+            Logger.log("Offline overflow disk drain complete ($totalDrained events)")
+            return totalDrained
+        } finally {
+            isDraining.set(false)
+        }
     }
 
     /**
-     * Check if the flush threshold has been reached (either event count or byte size).
+     * Write remaining events back to the disk store as a checkpoint.
+     * Used during drain for crash safety and on failure to preserve unsent events.
      */
-    @Synchronized
-    fun shouldFlushToDisk(): Boolean {
-        return memoryQueue.size >= flushThresholdEvents || estimatedBytes >= flushThresholdBytes
+    private fun writeCheckpoint(events: List<EnrichedEventPayload>) {
+        if (events.isEmpty()) return
+        synchronized(diskLock) {
+            diskStore.write(QueueSnapshot(version = 1, events = events))
+            hasOverflowData = true
+        }
     }
 
+
+    // ===== Internal =====
+
     /**
-     * Get the estimated total byte size of events in memory.
+     * Flush all events from memory queue to disk. Appends to any existing disk data,
+     * enforces [maxOfflineDiskEvents] cap, and resets the memory queue.
+     * Must be called while holding `synchronized(this)`.
+     *
+     * Note: performs disk I/O (read + write) while holding both `this` and `diskLock`.
+     * Local disk I/O is sub-millisecond for typical event sizes, so this is acceptable
+     * even on the enqueue hot path. Revisit if profiling shows contention.
      */
-    @Synchronized
-    fun estimatedSizeBytes(): Long = estimatedBytes
+    private fun flushMemoryToDiskInternal() {
+        if (memoryQueue.isEmpty()) return
 
-    private fun dropOldest(reason: String) {
-        memoryQueue.removeFirstOrNull()?.let { dropped ->
-            val size = eventSizes.removeFirstOrNull() ?: 0L
-            estimatedBytes -= size
-            Logger.warn("Queue $reason reached - dropped oldest event (messageId: ${dropped.messageId})")
+        val batch = memoryQueue.toList()
+        memoryQueue.clear()
+        eventSizes.clear()
+        estimatedBytes = 0L
+
+        synchronized(diskLock) {
+            val existing = try {
+                diskStore.read()?.events ?: emptyList()
+            } catch (e: Exception) {
+                Logger.warn("Failed to read disk store during flush: ${e.message}")
+                emptyList()
+            }
+
+            var combined = existing + batch
+            if (combined.size > maxOfflineDiskEvents) {
+                val dropCount = combined.size - maxOfflineDiskEvents
+                combined = combined.drop(dropCount)
+                Logger.warn("Offline disk cap reached — dropped $dropCount oldest events")
+            }
+
+            diskStore.write(QueueSnapshot(version = 1, events = combined))
+            hasOverflowData = true
+            Logger.log("Flushed ${batch.size} events to disk (total on disk: ${combined.size})")
         }
     }
 
+    /** Delete the disk file and clear the hasOverflowData flag. */
+    private fun deleteDiskStore() {
+        diskStore.delete()
+        hasOverflowData = false
+    }
+
     private fun estimateEventSize(event: EnrichedEventPayload): Long {
         return try {
             json.encodeToString(event).length.toLong()
diff --git a/metarouter-sdk/src/main/java/com/metarouter/analytics/storage/EventDiskStore.kt b/metarouter-sdk/src/main/java/com/metarouter/analytics/storage/EventDiskStore.kt
index 9c6b099..91cdaf4 100644
--- a/metarouter-sdk/src/main/java/com/metarouter/analytics/storage/EventDiskStore.kt
+++ b/metarouter-sdk/src/main/java/com/metarouter/analytics/storage/EventDiskStore.kt
@@ -69,6 +69,9 @@ class EventDiskStore(private val baseDir: File) {
     private val tmpFile: File
         get() = File(snapshotDir, "$SNAPSHOT_FILENAME.tmp")
 
+    /** Cheap check — does a snapshot file exist on disk? */
+    fun exists(): Boolean = snapshotFile.exists()
+
     /**
      * Write a snapshot to disk. Uses atomic write-to-tmp-then-rename.
      * Overwrites any existing snapshot completely.
diff --git a/metarouter-sdk/src/test/java/com/metarouter/analytics/MetaRouterAnalyticsClientTest.kt b/metarouter-sdk/src/test/java/com/metarouter/analytics/MetaRouterAnalyticsClientTest.kt
index 4780a43..0cc9d9e 100644
--- a/metarouter-sdk/src/test/java/com/metarouter/analytics/MetaRouterAnalyticsClientTest.kt
+++ b/metarouter-sdk/src/test/java/com/metarouter/analytics/MetaRouterAnalyticsClientTest.kt
@@ -420,7 +420,7 @@ class MetaRouterAnalyticsClientTest {
     // ===== Queue Overflow =====
 
     @Test
-    fun `queue overflow drops oldest events`() = runBlocking {
+    fun `queue overflow flushes to disk instead of dropping`() = runBlocking {
         val smallOptions = options.copy(maxQueueEvents = 5)
         val client = MetaRouterAnalyticsClient.initialize(context, smallOptions)
 
@@ -428,13 +428,13 @@ class MetaRouterAnalyticsClientTest {
         repeat(10) { i ->
             client.track("Event $i")
         }
-        // Wait until queue has processed and is at capacity (5 events)
-        awaitCondition { (client.getDebugInfo()["queueLength"] as Int) == 5 }
+        // Wait until all events have been processed through the channel
+        awaitCondition { (client.getDebugInfo()["queueLength"] as Int) > 0 }
 
         val debugInfo = client.getDebugInfo()
 
-        // Queue should have exactly 5 events (oldest dropped)
-        assertEquals(5, debugInfo["queueLength"])
+        // Queue should not exceed maxQueueEvents (overflow goes to disk)
+        assertTrue((debugInfo["queueLength"] as Int) <= 5)
     }
 
     // ===== Flush (Stub) =====
diff --git a/metarouter-sdk/src/test/java/com/metarouter/analytics/dispatcher/DispatcherTest.kt b/metarouter-sdk/src/test/java/com/metarouter/analytics/dispatcher/DispatcherTest.kt
index 86fa564..644f142 100644
--- a/metarouter-sdk/src/test/java/com/metarouter/analytics/dispatcher/DispatcherTest.kt
+++ b/metarouter-sdk/src/test/java/com/metarouter/analytics/dispatcher/DispatcherTest.kt
@@ -1050,6 +1050,72 @@ class DispatcherTest {
         dispatcher.stop()
     }
 
+    // ===== sendBatchDirect Tests =====
+
+    @Test
+    fun `sendBatchDirect returns 200 response on success`() = runTest {
+        val dispatcher = createDispatcher()
+        networkClient.nextResponse = NetworkResponse(200, emptyMap(), null)
+
+        val events = listOf(createEvent("direct-1"), createEvent("direct-2"))
+        val result = dispatcher.sendBatchDirect(events)
+
+        assertNotNull(result)
+        assertEquals(200, result!!.statusCode)
+        assertEquals(1, networkClient.requests.size)
+    }
+
+    @Test
+    fun `sendBatchDirect returns 500 response on server error`() = runTest {
+        val dispatcher = createDispatcher()
+        networkClient.nextResponse = NetworkResponse(500, emptyMap(), null)
+
+        val events = listOf(createEvent("direct-1"))
+        val result = dispatcher.sendBatchDirect(events)
+
+        assertNotNull(result)
+        assertEquals(500, result!!.statusCode)
+    }
+
+    @Test
+    fun `sendBatchDirect returns null on network error`() = runTest {
+        val dispatcher = createDispatcher()
+        networkClient.nextException = java.io.IOException("Connection refused")
+
+        val events = listOf(createEvent("direct-1"))
+        val result = dispatcher.sendBatchDirect(events)
+
+        assertNull(result)
+    }
+
+    @Test
+    fun `sendBatchDirect returns 200 for empty list`() = runTest {
+        val dispatcher = createDispatcher()
+
+        val result = dispatcher.sendBatchDirect(emptyList())
+
+        assertNotNull(result)
+        assertEquals(200, result!!.statusCode)
+        assertEquals(0, networkClient.requests.size) // No request made
+    }
+
+    @Test
+    fun `sendBatchDirect does not affect memory queue`() = runTest {
+        val dispatcher = createDispatcher()
+        networkClient.nextResponse = NetworkResponse(200, emptyMap(), null)
+
+        // Add events to memory queue
+        queue.enqueue(createEvent("queue-1"))
+        assertEquals(1, queue.size())
+
+        // Send different events via direct path
+        val directEvents = listOf(createEvent("direct-1"))
+        dispatcher.sendBatchDirect(directEvents)
+
+        // Memory queue should be untouched
+        assertEquals(1, queue.size())
+    }
+
     // ===== Helper Methods =====
 
     private fun TestScope.createDispatcher(): Dispatcher {
diff --git a/metarouter-sdk/src/test/java/com/metarouter/analytics/queue/PersistableEventQueueTest.kt b/metarouter-sdk/src/test/java/com/metarouter/analytics/queue/PersistableEventQueueTest.kt
index 50d20a2..ff3ba14 100644
--- a/metarouter-sdk/src/test/java/com/metarouter/analytics/queue/PersistableEventQueueTest.kt
+++ b/metarouter-sdk/src/test/java/com/metarouter/analytics/queue/PersistableEventQueueTest.kt
@@ -1,8 +1,20 @@
 package com.metarouter.analytics.queue
 
+import android.util.Log
+import com.metarouter.analytics.InitOptions
+import com.metarouter.analytics.dispatcher.Dispatcher
+import com.metarouter.analytics.dispatcher.DispatcherConfig
+import com.metarouter.analytics.network.CircuitBreaker
+import com.metarouter.analytics.network.FakeNetworkClient
+import com.metarouter.analytics.network.NetworkResponse
 import com.metarouter.analytics.storage.EventDiskStore
 import com.metarouter.analytics.storage.QueueSnapshot
 import com.metarouter.analytics.types.*
+import io.mockk.every
+import io.mockk.mockkStatic
+import io.mockk.unmockkStatic
+import kotlinx.coroutines.async
+import kotlinx.coroutines.test.runTest
 import org.junit.After
 import org.junit.Assert.*
 import org.junit.Before
@@ -57,7 +69,7 @@ class PersistableEventQueueTest {
     }
 
     @Test
-    fun `enqueue drops oldest when event count capacity exceeded`() {
+    fun `enqueue flushes to disk when event count capacity exceeded`() {
         val smallQueue = PersistableEventQueue(
             maxCapacity = 5,
             diskStore = diskStore,
@@ -70,13 +82,15 @@ class PersistableEventQueueTest {
             smallQueue.enqueue(createTestEvent("msg-$i"))
         }
 
-        assertEquals(5, smallQueue.size())
-        val drained = smallQueue.drain(5)
-        assertEquals("msg-1", drained[0].messageId)
+        // Memory queue has just the latest event; others went to disk
+        assertEquals(1, smallQueue.size())
+        val snapshot = diskStore.read()
+        assertNotNull(snapshot)
+        assertEquals(5, snapshot!!.events.size)
     }
 
     @Test
-    fun `enqueue drops oldest when byte capacity exceeded`() {
+    fun `enqueue flushes to disk when byte capacity exceeded`() {
         val tinyQueue = PersistableEventQueue(
             maxCapacity = 2000,
             diskStore = diskStore,
@@ -89,8 +103,9 @@ class PersistableEventQueueTest {
             tinyQueue.enqueue(createTestEvent("msg-$i"))
         }
 
-        assertTrue(tinyQueue.size() < 10)
+        // Events that exceeded byte cap got flushed to disk
         assertTrue(tinyQueue.estimatedSizeBytes() <= 500)
+        assertTrue(tinyQueue.hasOverflowData())
     }
 
     // ===== Drain (memory-only) =====
@@ -122,13 +137,14 @@ class PersistableEventQueueTest {
     // ===== Flush to Disk =====
 
     @Test
-    fun `flushToDisk writes current memory state to disk`() {
+    fun `flushToDisk writes memory state to disk and clears memory`() {
         repeat(3) { i ->
             queue.enqueue(createTestEvent("msg-$i"))
         }
 
         queue.flushToDisk()
 
+        assertEquals(0, queue.size())
         val snapshot = diskStore.read()
         assertNotNull(snapshot)
         assertEquals(3, snapshot!!.events.size)
@@ -136,35 +152,37 @@ class PersistableEventQueueTest {
     }
 
     @Test
-    fun `flushToDisk fully overwrites previous snapshot`() {
+    fun `flushToDisk appends memory to existing disk contents and clears memory`() {
         queue.enqueue(createTestEvent("msg-1"))
         queue.flushToDisk()
+        // Memory queue cleared, msg-1 is on disk
+        assertEquals(0, queue.size())
 
-        queue.drain(1)
         queue.enqueue(createTestEvent("msg-2"))
         queue.enqueue(createTestEvent("msg-3"))
         queue.flushToDisk()
 
         val snapshot = diskStore.read()
         assertNotNull(snapshot)
-        assertEquals(2, snapshot!!.events.size)
-        assertEquals("msg-2", snapshot.events[0].messageId)
-        assertEquals("msg-3", snapshot.events[1].messageId)
+        assertEquals(3, snapshot!!.events.size)
+        assertEquals("msg-1", snapshot.events[0].messageId)
+        assertEquals("msg-2", snapshot.events[1].messageId)
+        assertEquals("msg-3", snapshot.events[2].messageId)
     }
 
     @Test
-    fun `flushToDisk with empty queue skips write and deletes existing snapshot`() {
+    fun `flushToDisk with empty queue is a no-op`() {
         // Write a snapshot with events first
         queue.enqueue(createTestEvent("msg-1"))
         queue.flushToDisk()
-        assertNotNull(diskStore.read()) // snapshot exists
+        assertNotNull(diskStore.read())
 
-        // Drain all events, then flush empty queue
-        queue.drain(1)
+        // Memory is now empty; flushing again shouldn't delete or modify disk
         queue.flushToDisk()
 
-        // Existing snapshot should be deleted, not overwritten with empty
-        assertNull(diskStore.read())
+        val snapshot = diskStore.read()
+        assertNotNull(snapshot)
+        assertEquals(1, snapshot!!.events.size)
     }
 
     @Test
@@ -211,172 +229,41 @@ class PersistableEventQueueTest {
     }
 
     // ===== Rehydration =====
+    // With consolidated disk store, events stay on disk and drain to network.
+    // Memory queue starts empty on rehydrate. TTL filtering happens in drain, not rehydrate.
 
     @Test
-    fun `rehydrate loads events from disk into memory`() {
+    fun `rehydrate sets hasOverflowData when disk file exists`() {
         val events = listOf(createTestEvent("disk-1"), createTestEvent("disk-2"))
-        val snapshot = QueueSnapshot(version = 1, events = events)
-        diskStore.write(snapshot)
-
-        queue.rehydrate()
-
-        assertEquals(2, queue.size())
-        val drained = queue.drain(10)
-        assertEquals("disk-1", drained[0].messageId)
-        assertEquals("disk-2", drained[1].messageId)
-    }
-
-    @Test
-    fun `rehydrate only happens once per process`() {
-        val events = listOf(createTestEvent("disk-1"))
         diskStore.write(QueueSnapshot(version = 1, events = events))
 
-        queue.rehydrate()
-        assertEquals(1, queue.size())
-
-        queue.drain(1)
-        val queue2 = PersistableEventQueue(
+        val freshQueue = PersistableEventQueue(
             maxCapacity = 2000,
             diskStore = diskStore,
-            flushThresholdEvents = 500,
-            flushThresholdBytes = 2 * 1024 * 1024,
-            maxCapacityBytes = 5 * 1024 * 1024
+            eventTTLMs = 0
         )
+        freshQueue.rehydrate()
 
-        diskStore.write(QueueSnapshot(version = 1, events = listOf(createTestEvent("disk-2"))))
-
-        queue2.rehydrate()
-        assertEquals(0, queue2.size())
+        // Memory queue starts empty; disk file preserved for drain
+        assertEquals(0, freshQueue.size())
+        assertTrue(freshQueue.hasOverflowData())
+        assertNotNull(diskStore.read())
     }
 
     @Test
-    fun `rehydrate with no disk file does nothing`() {
+    fun `rehydrate with no disk file sets hasOverflowData false`() {
         queue.rehydrate()
         assertEquals(0, queue.size())
+        assertFalse(queue.hasOverflowData())
     }
 
     @Test
-    fun `rehydrate drops oldest events if disk exceeds capacity`() {
-        val smallCapQueue = PersistableEventQueue(
-            maxCapacity = 3,
-            diskStore = diskStore,
-            flushThresholdEvents = 500,
-            flushThresholdBytes = 2 * 1024 * 1024,
-            maxCapacityBytes = 5 * 1024 * 1024,
-            eventTTLMs = 0
-        )
-
-        val events = (1..5).map { createTestEvent("msg-$it") }
-        diskStore.write(QueueSnapshot(version = 1, events = events))
-
-        smallCapQueue.rehydrate()
-
-        assertEquals(3, smallCapQueue.size())
-        val drained = smallCapQueue.drain(10)
-        assertEquals("msg-3", drained[0].messageId)
-        assertEquals("msg-4", drained[1].messageId)
-        assertEquals("msg-5", drained[2].messageId)
-    }
-
-    @Test
-    fun `rehydrate deletes disk file after loading`() {
-        val events = listOf(createTestEvent("disk-1"))
-        diskStore.write(QueueSnapshot(version = 1, events = events))
-
+    fun `rehydrate only sets rehydrated flag once per process`() {
+        PersistableEventQueue.resetRehydrationFlag()
         queue.rehydrate()
-
-        assertNull(diskStore.read())
-    }
-
-    // ===== Event TTL =====
-
-    @Test
-    fun `rehydrate drops events older than TTL`() {
-        val ttlQueue = PersistableEventQueue(
-            maxCapacity = 2000,
-            diskStore = diskStore,
-            flushThresholdEvents = 500,
-            flushThresholdBytes = 2 * 1024 * 1024,
-            maxCapacityBytes = 5 * 1024 * 1024,
-            eventTTLMs = 7L * 24 * 60 * 60 * 1000 // 7 days
-        )
-
-        val now = System.currentTimeMillis()
-        val eightDaysAgo = formatTimestamp(now - 8L * 24 * 60 * 60 * 1000)
-        val twoDaysAgo = formatTimestamp(now - 2L * 24 * 60 * 60 * 1000)
-        val recent = formatTimestamp(now - 60_000) // 1 minute ago
-
-        val events = listOf(
-            createTestEvent("old-1", timestamp = eightDaysAgo),
-            createTestEvent("mid-1", timestamp = twoDaysAgo),
-            createTestEvent("new-1", timestamp = recent)
-        )
-        diskStore.write(QueueSnapshot(version = 1, events = events))
-
-        ttlQueue.rehydrate()
-
-        assertEquals(2, ttlQueue.size())
-        val drained = ttlQueue.drain(10)
-        assertEquals("mid-1", drained[0].messageId)
-        assertEquals("new-1", drained[1].messageId)
-    }
-
-    @Test
-    fun `rehydrate TTL filter runs before capacity trim`() {
-        val ttlQueue = PersistableEventQueue(
-            maxCapacity = 2,
-            diskStore = diskStore,
-            flushThresholdEvents = 500,
-            flushThresholdBytes = 2 * 1024 * 1024,
-            maxCapacityBytes = 5 * 1024 * 1024,
-            eventTTLMs = 7L * 24 * 60 * 60 * 1000
-        )
-
-        val now = System.currentTimeMillis()
-        val eightDaysAgo = formatTimestamp(now - 8L * 24 * 60 * 60 * 1000)
-        val twoDaysAgo = formatTimestamp(now - 2L * 24 * 60 * 60 * 1000)
-        val oneDayAgo = formatTimestamp(now - 1L * 24 * 60 * 60 * 1000)
-        val recent = formatTimestamp(now - 60_000)
-
-        // 4 events: 2 expired, 2 valid. maxCapacity = 2.
-        // TTL filter removes the 2 expired. 2 valid remain. Capacity is satisfied.
-        val events = listOf(
-            createTestEvent("expired-1", timestamp = eightDaysAgo),
-            createTestEvent("expired-2", timestamp = eightDaysAgo),
-            createTestEvent("valid-1", timestamp = twoDaysAgo),
-            createTestEvent("valid-2", timestamp = oneDayAgo)
-        )
-        diskStore.write(QueueSnapshot(version = 1, events = events))
-
-        ttlQueue.rehydrate()
-
-        assertEquals(2, ttlQueue.size())
-        val drained = ttlQueue.drain(10)
-        assertEquals("valid-1", drained[0].messageId)
-        assertEquals("valid-2", drained[1].messageId)
-    }
-
-    @Test
-    fun `rehydrate with unparseable timestamp keeps event`() {
-        val ttlQueue = PersistableEventQueue(
-            maxCapacity = 2000,
-            diskStore = diskStore,
-            flushThresholdEvents = 500,
-            flushThresholdBytes = 2 * 1024 * 1024,
-            maxCapacityBytes = 5 * 1024 * 1024,
-            eventTTLMs = 7L * 24 * 60 * 60 * 1000
-        )
-
-        val events = listOf(
-            createTestEvent("bad-ts", timestamp = "not-a-timestamp"),
-            createTestEvent("good-ts", timestamp = formatTimestamp(System.currentTimeMillis() - 60_000))
-        )
-        diskStore.write(QueueSnapshot(version = 1, events = events))
-
-        ttlQueue.rehydrate()
-
-        // Unparseable timestamp should be kept (fail-open), not dropped
-        assertEquals(2, ttlQueue.size())
+        // Second call is a no-op (guarded by hasRehydrated flag)
+        queue.rehydrate()
+        // No assertions — just confirming it doesn't throw
     }
 
     // ===== Clear =====
@@ -450,37 +337,7 @@ class PersistableEventQueueTest {
     // ===== End-to-end Scenarios =====
 
     @Test
-    fun `full lifecycle - enqueue, background flush, process restart, rehydrate`() {
-        // Simulate first session
-        repeat(10) { i ->
-            queue.enqueue(createTestEvent("session1-msg-$i"))
-        }
-        assertEquals(10, queue.size())
-
-        // App backgrounds -> flush to disk
-        queue.flushToDisk()
-
-        // Simulate process restart: create new queue with same disk store
-        PersistableEventQueue.resetRehydrationFlag()
-        val queue2 = PersistableEventQueue(
-            maxCapacity = 2000,
-            diskStore = diskStore,
-            flushThresholdEvents = 500,
-            flushThresholdBytes = 2 * 1024 * 1024,
-            maxCapacityBytes = 5 * 1024 * 1024,
-            eventTTLMs = 0
-        )
-
-        queue2.rehydrate()
-
-        assertEquals(10, queue2.size())
-        val drained = queue2.drain(10)
-        assertEquals("session1-msg-0", drained[0].messageId)
-        assertEquals("session1-msg-9", drained[9].messageId)
-    }
-
-    @Test
-    fun `events drained before background are not in snapshot`() {
+    fun `events drained before background are not flushed to disk`() {
         repeat(5) { i ->
             queue.enqueue(createTestEvent("msg-$i"))
         }
@@ -509,74 +366,647 @@ class PersistableEventQueueTest {
         assertFalse(queue.shouldFlushToDisk())
     }
 
+    // ===== Offline Overflow =====
+
     @Test
-    fun `multiple flush-rehydrate cycles preserve data correctly`() {
-        // Cycle 1: enqueue and flush
-        queue.enqueue(createTestEvent("cycle1-a"))
-        queue.enqueue(createTestEvent("cycle1-b"))
-        queue.flushToDisk()
+    fun `capacity overflow flushes entire queue to disk`() {
 
-        // Cycle 2: restart, rehydrate, add more, flush again
-        PersistableEventQueue.resetRehydrationFlag()
-        val queue2 = PersistableEventQueue(
-            maxCapacity = 2000,
+        val smallQueue = PersistableEventQueue(
+            maxCapacity = 5,
             diskStore = diskStore,
-            flushThresholdEvents = 500,
-            flushThresholdBytes = 2 * 1024 * 1024,
-            maxCapacityBytes = 5 * 1024 * 1024,
+            maxOfflineDiskEvents = 100,
             eventTTLMs = 0
         )
-        queue2.rehydrate()
-        queue2.enqueue(createTestEvent("cycle2-c"))
-        queue2.flushToDisk()
 
-        // Cycle 3: restart, rehydrate
-        PersistableEventQueue.resetRehydrationFlag()
-        val queue3 = PersistableEventQueue(
-            maxCapacity = 2000,
+        // Fill queue to capacity
+        repeat(5) { i -> smallQueue.enqueue(createTestEvent("fill-$i")) }
+        assertEquals(5, smallQueue.size())
+
+        // Enqueue one more — should flush all 5 to disk, then add the new event
+        smallQueue.enqueue(createTestEvent("overflow-0"))
+
+        // Memory queue should have just the new event
+        assertEquals(1, smallQueue.size())
+
+        // All 5 original events should be on overflow disk
+        val snapshot = diskStore.read()
+        assertNotNull(snapshot)
+        assertEquals(5, snapshot!!.events.size)
+        assertEquals("fill-0", snapshot.events[0].messageId)
+        assertEquals("fill-4", snapshot.events[4].messageId)
+
+        // Drain memory to verify contents
+        val drained = smallQueue.drain(10)
+        assertEquals(1, drained.size)
+        assertEquals("overflow-0", drained[0].messageId)
+
+    }
+
+    @Test
+    fun `flushToOfflineStorage drains entire queue to disk`() {
+
+        val smallQueue = PersistableEventQueue(
+            maxCapacity = 5,
             diskStore = diskStore,
-            flushThresholdEvents = 500,
-            flushThresholdBytes = 2 * 1024 * 1024,
-            maxCapacityBytes = 5 * 1024 * 1024,
+            maxOfflineDiskEvents = 100,
             eventTTLMs = 0
         )
-        queue3.rehydrate()
 
-        assertEquals(3, queue3.size())
-        val drained = queue3.drain(10)
-        assertEquals("cycle1-a", drained[0].messageId)
-        assertEquals("cycle1-b", drained[1].messageId)
-        assertEquals("cycle2-c", drained[2].messageId)
+        // Fill queue with 3 events
+        repeat(3) { i -> smallQueue.enqueue(createTestEvent("event-$i")) }
+        assertEquals(3, smallQueue.size())
+
+        // Flush to offline storage (simulates dispatcher offline flush)
+        val flushed = smallQueue.flushToOfflineStorage()
+        assertTrue(flushed)
+
+        // Memory queue should be empty
+        assertEquals(0, smallQueue.size())
+
+        // Events should be on overflow disk
+        val snapshot = diskStore.read()
+        assertNotNull(snapshot)
+        assertEquals(3, snapshot!!.events.size)
+
     }
 
     @Test
-    fun `flush overwrites previous state - no duplicates`() {
-        queue.enqueue(createTestEvent("batch1-a"))
-        queue.enqueue(createTestEvent("batch1-b"))
-        queue.flushToDisk()
+    fun `flushToOfflineStorage appends to existing overflow disk data`() {
 
-        // Drain all, add new events
-        queue.drain(10)
-        queue.enqueue(createTestEvent("batch2-a"))
-        queue.flushToDisk()
+        val smallQueue = PersistableEventQueue(
+            maxCapacity = 5,
+            diskStore = diskStore,
+            maxOfflineDiskEvents = 100,
+            eventTTLMs = 0
+        )
+
+        // Pre-populate overflow disk with 3 events
+        val existing = (1..3).map { createTestEvent("existing-$it") }
+        diskStore.write(QueueSnapshot(version = 1, events = existing))
+
+        // Add 2 events to memory and flush
+        repeat(2) { i -> smallQueue.enqueue(createTestEvent("new-$i")) }
+        smallQueue.flushToOfflineStorage()
+
+        // Disk should have all 5
+        val snapshot = diskStore.read()
+        assertNotNull(snapshot)
+        assertEquals(5, snapshot!!.events.size)
+        assertEquals("existing-1", snapshot.events[0].messageId)
+        assertEquals("new-1", snapshot.events[4].messageId)
+
+    }
+
+    @Test
+    fun `offline disk overflow respects maxOfflineDiskEvents cap`() {
+
+        val smallQueue = PersistableEventQueue(
+            maxCapacity = 5,
+            diskStore = diskStore,
+            maxOfflineDiskEvents = 8,
+            eventTTLMs = 0
+        )
+
+        // Fill and overflow multiple times to exceed cap
+        // Round 1: fill 5, overflow triggers flush of 5 to disk
+        repeat(5) { i -> smallQueue.enqueue(createTestEvent("batch1-$i")) }
+        smallQueue.flushToOfflineStorage()
+
+        // Round 2: fill 5 more, flush again — total would be 10, cap is 8
+        repeat(5) { i -> smallQueue.enqueue(createTestEvent("batch2-$i")) }
+        smallQueue.flushToOfflineStorage()
+
+        val snapshot = diskStore.read()
+        assertNotNull(snapshot)
+        assertTrue("Should cap at maxOfflineDiskEvents", snapshot!!.events.size <= 8)
+
+    }
+
+    @Test
+    fun `drainDiskOverflowToNetwork sends directly to network`() = runTest {
+        mockkStatic(Log::class)
+        every { Log.d(any(), any()) } returns 0
+        every { Log.w(any(), any<String>()) } returns 0
+        every { Log.e(any(), any()) } returns 0
+        every { Log.e(any(), any(), any()) } returns 0
+
+
+        val overflowQueue = PersistableEventQueue(
+            maxCapacity = 5,
+            diskStore = diskStore,
+            maxOfflineDiskEvents = 100,
+            eventTTLMs = 0
+        )
+
+        // Pre-populate overflow disk
+        val events = (1..5).map { createTestEvent("overflow-$it") }
+        diskStore.write(QueueSnapshot(version = 1, events = events))
+
+        // Set up dispatcher with fake network
+        val fakeClient = FakeNetworkClient()
+        fakeClient.nextResponse = NetworkResponse(200, emptyMap(), null)
+        val initOptions = InitOptions(
+            writeKey = "test-key",
+            ingestionHost = "https://api.example.com",
+            flushIntervalSeconds = 10
+        )
+        val dispatcher = Dispatcher(
+            options = initOptions,
+            queue = overflowQueue,
+            networkClient = fakeClient,
+            circuitBreaker = CircuitBreaker(),
+            scope = this,
+            config = DispatcherConfig()
+        )
+
+        val drained = overflowQueue.drainDiskOverflowToNetwork(dispatcher)
+
+        assertEquals(5, drained)
+        assertEquals(1, fakeClient.requests.size)
+        assertNull(diskStore.read())
+        assertEquals(0, overflowQueue.size())
+
+        unmockkStatic(Log::class)
+    }
+
+    @Test
+    fun `drainDiskOverflowToNetwork stops on network failure`() = runTest {
+        mockkStatic(Log::class)
+        every { Log.d(any(), any()) } returns 0
+        every { Log.w(any(), any<String>()) } returns 0
+        every { Log.e(any(), any()) } returns 0
+        every { Log.e(any(), any(), any()) } returns 0
 
+
+        val overflowQueue = PersistableEventQueue(
+            maxCapacity = 5,
+            diskStore = diskStore,
+            maxOfflineDiskEvents = 100,
+            eventTTLMs = 0
+        )
+
+        // Pre-populate overflow disk
+        val events = (1..5).map { createTestEvent("overflow-$it") }
+        diskStore.write(QueueSnapshot(version = 1, events = events))
+
+        // Set up dispatcher that fails
+        val fakeClient = FakeNetworkClient()
+        fakeClient.nextResponse = NetworkResponse(500, emptyMap(), null)
+        val initOptions = InitOptions(
+            writeKey = "test-key",
+            ingestionHost = "https://api.example.com",
+            flushIntervalSeconds = 10
+        )
+        val dispatcher = Dispatcher(
+            options = initOptions,
+            queue = overflowQueue,
+            networkClient = fakeClient,
+            circuitBreaker = CircuitBreaker(),
+            scope = this,
+            config = DispatcherConfig()
+        )
+
+        val drained = overflowQueue.drainDiskOverflowToNetwork(dispatcher)
+
+        assertEquals(0, drained)
+        val remaining = diskStore.read()
+        assertNotNull(remaining)
+        assertEquals(5, remaining!!.events.size)
+
+        unmockkStatic(Log::class)
+    }
+
+    @Test
+    fun `clear deletes overflow disk file`() {
+
+        val overflowQueue = PersistableEventQueue(
+            maxCapacity = 5,
+            diskStore = diskStore,
+            maxOfflineDiskEvents = 100,
+            eventTTLMs = 0
+        )
+
+        diskStore.write(QueueSnapshot(version = 1, events = listOf(createTestEvent("overflow-1"))))
+        overflowQueue.enqueue(createTestEvent("memory-1"))
+
+        overflowQueue.clear()
+
+        assertEquals(0, overflowQueue.size())
+        assertNull(diskStore.read())
+        assertNull(diskStore.read())
+
+    }
+
+    @Test
+    fun `app killed while offline, relaunch online, disk overflow drains to network`() = runTest {
+        mockkStatic(Log::class)
+        every { Log.d(any(), any()) } returns 0
+        every { Log.w(any(), any<String>()) } returns 0
+        every { Log.e(any(), any()) } returns 0
+        every { Log.e(any(), any(), any()) } returns 0
+
+
+        // Simulate previous session: overflow events were written to disk
+        val previousEvents = (1..3).map { createTestEvent("prev-session-$it") }
+        diskStore.write(QueueSnapshot(version = 1, events = previousEvents))
+
+        // New session — create fresh queue (simulating app relaunch)
         PersistableEventQueue.resetRehydrationFlag()
-        val queue2 = PersistableEventQueue(
+        val freshQueue = PersistableEventQueue(
             maxCapacity = 2000,
             diskStore = diskStore,
-            flushThresholdEvents = 500,
-            flushThresholdBytes = 2 * 1024 * 1024,
-            maxCapacityBytes = 5 * 1024 * 1024,
+            maxOfflineDiskEvents = 100,
             eventTTLMs = 0
         )
-        queue2.rehydrate()
+        freshQueue.rehydrate()
 
-        // Should only have batch2 events — full overwrite, no duplicates
-        assertEquals(1, queue2.size())
-        val drained = queue2.drain(10)
-        assertEquals("batch2-a", drained[0].messageId)
+        // Set up dispatcher with working network (online at launch)
+        val fakeClient = FakeNetworkClient()
+        fakeClient.nextResponse = NetworkResponse(200, emptyMap(), null)
+        val initOptions = InitOptions(
+            writeKey = "test-key",
+            ingestionHost = "https://api.example.com",
+            flushIntervalSeconds = 10
+        )
+        val dispatcher = Dispatcher(
+            options = initOptions,
+            queue = freshQueue,
+            networkClient = fakeClient,
+            circuitBreaker = CircuitBreaker(),
+            scope = this,
+            config = DispatcherConfig()
+        )
+
+        val drained = freshQueue.drainDiskOverflowToNetwork(dispatcher)
+
+        assertEquals(3, drained)
+        assertEquals(1, fakeClient.requests.size)
+        assertNull(diskStore.read())
+
+        unmockkStatic(Log::class)
     }
 
+    @Test
+    fun `flushToOfflineStorage returns false when memory queue is empty`() {
+        val smallQueue = PersistableEventQueue(
+            maxCapacity = 5,
+            diskStore = diskStore,
+            eventTTLMs = 0
+        )
+
+        // No events in memory queue
+        assertFalse(smallQueue.flushToOfflineStorage())
+        assertEquals(0, smallQueue.size())
+    }
+
+    // ===== Drain Response Handling =====
+
+    @Test
+    fun `drain halves batch on 413 and retries`() = runTest {
+        mockkStatic(Log::class)
+        every { Log.d(any(), any()) } returns 0
+        every { Log.w(any(), any<String>()) } returns 0
+        every { Log.e(any(), any()) } returns 0
+        every { Log.e(any(), any(), any()) } returns 0
+
+
+        val overflowQueue = PersistableEventQueue(
+            maxCapacity = 5,
+            diskStore = diskStore,
+            maxOfflineDiskEvents = 100,
+            eventTTLMs = 0
+        )
+
+        // Pre-populate overflow disk with 4 events
+        val events = (1..4).map { createTestEvent("overflow-$it") }
+        diskStore.write(QueueSnapshot(version = 1, events = events))
+
+        val fakeClient = FakeNetworkClient()
+        // First call: 413, second call (halved batch): 200, third call: 200
+        fakeClient.enqueueResponses(
+            NetworkResponse(413, emptyMap(), null),
+            NetworkResponse(200, emptyMap(), null),
+            NetworkResponse(200, emptyMap(), null)
+        )
+        val dispatcher = Dispatcher(
+            options = InitOptions(writeKey = "test-key", ingestionHost = "https://api.example.com", flushIntervalSeconds = 10),
+            queue = overflowQueue,
+            networkClient = fakeClient,
+            circuitBreaker = CircuitBreaker(),
+            scope = this,
+            config = DispatcherConfig()
+        )
+
+        val drained = overflowQueue.drainDiskOverflowToNetwork(dispatcher)
+
+        assertEquals(4, drained)
+        assertNull(diskStore.read())
+
+        unmockkStatic(Log::class)
+    }
+
+    @Test
+    fun `drain drops oversized event at batchSize 1 on 413`() = runTest {
+        mockkStatic(Log::class)
+        every { Log.d(any(), any()) } returns 0
+        every { Log.w(any(), any<String>()) } returns 0
+        every { Log.e(any(), any()) } returns 0
+        every { Log.e(any(), any(), any()) } returns 0
+
+
+        val overflowQueue = PersistableEventQueue(
+            maxCapacity = 5,
+            diskStore = diskStore,
+            maxOfflineDiskEvents = 100,
+            eventTTLMs = 0
+        )
+
+        // 2 events — first will be too large, second should still send
+        val events = listOf(createTestEvent("big-event"), createTestEvent("normal-event"))
+        diskStore.write(QueueSnapshot(version = 1, events = events))
+
+        val fakeClient = FakeNetworkClient()
+        // 413 responses until batch is 1, then 413 again (drop), then 200 for remaining
+        fakeClient.enqueueResponses(
+            NetworkResponse(413, emptyMap(), null),  // batch=100 -> halve
+            NetworkResponse(413, emptyMap(), null),  // batch=50 -> halve (and so on until 1)
+            NetworkResponse(413, emptyMap(), null),
+            NetworkResponse(413, emptyMap(), null),
+            NetworkResponse(413, emptyMap(), null),
+            NetworkResponse(413, emptyMap(), null),
+            NetworkResponse(413, emptyMap(), null),  // batch=1 -> drop big-event
+            NetworkResponse(200, emptyMap(), null)   // normal-event succeeds
+        )
+        val dispatcher = Dispatcher(
+            options = InitOptions(writeKey = "test-key", ingestionHost = "https://api.example.com", flushIntervalSeconds = 10),
+            queue = overflowQueue,
+            networkClient = fakeClient,
+            circuitBreaker = CircuitBreaker(),
+            scope = this,
+            config = DispatcherConfig()
+        )
+
+        val drained = overflowQueue.drainDiskOverflowToNetwork(dispatcher)
+
+        // Only normal-event was successfully sent
+        assertEquals(1, drained)
+        assertNull(diskStore.read())
+
+        unmockkStatic(Log::class)
+    }
+
+    @Test
+    fun `drain deletes overflow store on fatal config error`() = runTest {
+        mockkStatic(Log::class)
+        every { Log.d(any(), any()) } returns 0
+        every { Log.w(any(), any<String>()) } returns 0
+        every { Log.e(any(), any()) } returns 0
+        every { Log.e(any(), any(), any()) } returns 0
+
+
+        val overflowQueue = PersistableEventQueue(
+            maxCapacity = 5,
+            diskStore = diskStore,
+            maxOfflineDiskEvents = 100,
+            eventTTLMs = 0
+        )
+
+        val events = (1..5).map { createTestEvent("overflow-$it") }
+        diskStore.write(QueueSnapshot(version = 1, events = events))
+
+        val fakeClient = FakeNetworkClient()
+        fakeClient.nextResponse = NetworkResponse(401, emptyMap(), null)
+        val dispatcher = Dispatcher(
+            options = InitOptions(writeKey = "test-key", ingestionHost = "https://api.example.com", flushIntervalSeconds = 10),
+            queue = overflowQueue,
+            networkClient = fakeClient,
+            circuitBreaker = CircuitBreaker(),
+            scope = this,
+            config = DispatcherConfig()
+        )
+
+        val drained = overflowQueue.drainDiskOverflowToNetwork(dispatcher)
+
+        assertEquals(0, drained)
+        assertNull(diskStore.read()) // Overflow store deleted on fatal error
+
+        unmockkStatic(Log::class)
+    }
+
+    @Test
+    fun `drain drops batch on client error and continues`() = runTest {
+        mockkStatic(Log::class)
+        every { Log.d(any(), any()) } returns 0
+        every { Log.w(any(), any<String>()) } returns 0
+        every { Log.e(any(), any()) } returns 0
+        every { Log.e(any(), any(), any()) } returns 0
+
+
+        val overflowQueue = PersistableEventQueue(
+            maxCapacity = 5,
+            diskStore = diskStore,
+            maxOfflineDiskEvents = 100,
+            eventTTLMs = 0
+        )
+
+        // Write 150 events so drain takes 2 batches (100 + 50)
+        val events = (1..150).map { createTestEvent("overflow-$it") }
+        diskStore.write(QueueSnapshot(version = 1, events = events))
+
+        val fakeClient = FakeNetworkClient()
+        // First batch: 400 (drop), second batch: 200 (success)
+        fakeClient.enqueueResponses(
+            NetworkResponse(400, emptyMap(), null),
+            NetworkResponse(200, emptyMap(), null)
+        )
+        val dispatcher = Dispatcher(
+            options = InitOptions(writeKey = "test-key", ingestionHost = "https://api.example.com", flushIntervalSeconds = 10),
+            queue = overflowQueue,
+            networkClient = fakeClient,
+            circuitBreaker = CircuitBreaker(),
+            scope = this,
+            config = DispatcherConfig()
+        )
+
+        val drained = overflowQueue.drainDiskOverflowToNetwork(dispatcher)
+
+        // Only second batch of 50 counted as drained
+        assertEquals(50, drained)
+        assertNull(diskStore.read())
+
+        unmockkStatic(Log::class)
+    }
+
+    @Test
+    fun `drain pauses on 429 rate limit`() = runTest {
+        mockkStatic(Log::class)
+        every { Log.d(any(), any()) } returns 0
+        every { Log.w(any(), any<String>()) } returns 0
+        every { Log.e(any(), any()) } returns 0
+        every { Log.e(any(), any(), any()) } returns 0
+
+
+        val overflowQueue = PersistableEventQueue(
+            maxCapacity = 5,
+            diskStore = diskStore,
+            maxOfflineDiskEvents = 100,
+            eventTTLMs = 0
+        )
+
+        val events = (1..5).map { createTestEvent("overflow-$it") }
+        diskStore.write(QueueSnapshot(version = 1, events = events))
+
+        val fakeClient = FakeNetworkClient()
+        fakeClient.nextResponse = NetworkResponse(429, emptyMap(), null)
+        val dispatcher = Dispatcher(
+            options = InitOptions(writeKey = "test-key", ingestionHost = "https://api.example.com", flushIntervalSeconds = 10),
+            queue = overflowQueue,
+            networkClient = fakeClient,
+            circuitBreaker = CircuitBreaker(),
+            scope = this,
+            config = DispatcherConfig()
+        )
+
+        val drained = overflowQueue.drainDiskOverflowToNetwork(dispatcher)
+
+        assertEquals(0, drained)
+        // Events should still be on disk
+        val remaining = diskStore.read()
+        assertNotNull(remaining)
+        assertEquals(5, remaining!!.events.size)
+
+        unmockkStatic(Log::class)
+    }
+
+    @Test
+    fun `drain filters expired events by TTL`() = runTest {
+        mockkStatic(Log::class)
+        every { Log.d(any(), any()) } returns 0
+        every { Log.w(any(), any<String>()) } returns 0
+        every { Log.e(any(), any()) } returns 0
+        every { Log.e(any(), any(), any()) } returns 0
+
+
+        val oneDayMs = 24L * 60 * 60 * 1000
+        val overflowQueue = PersistableEventQueue(
+            maxCapacity = 5,
+            diskStore = diskStore,
+            maxOfflineDiskEvents = 100,
+            eventTTLMs = oneDayMs // 1 day TTL
+        )
+
+        // Mix of expired and fresh events
+        val expiredTimestamp = formatTimestamp(System.currentTimeMillis() - 2 * oneDayMs) // 2 days ago
+        val freshTimestamp = formatTimestamp(System.currentTimeMillis() - 1000) // 1 second ago
+        val events = listOf(
+            createTestEvent("expired-1", timestamp = expiredTimestamp),
+            createTestEvent("expired-2", timestamp = expiredTimestamp),
+            createTestEvent("fresh-1", timestamp = freshTimestamp),
+            createTestEvent("fresh-2", timestamp = freshTimestamp)
+        )
+        diskStore.write(QueueSnapshot(version = 1, events = events))
+
+        val fakeClient = FakeNetworkClient()
+        fakeClient.nextResponse = NetworkResponse(200, emptyMap(), null)
+        val dispatcher = Dispatcher(
+            options = InitOptions(writeKey = "test-key", ingestionHost = "https://api.example.com", flushIntervalSeconds = 10),
+            queue = overflowQueue,
+            networkClient = fakeClient,
+            circuitBreaker = CircuitBreaker(),
+            scope = this,
+            config = DispatcherConfig()
+        )
+
+        val drained = overflowQueue.drainDiskOverflowToNetwork(dispatcher)
+
+        // Only the 2 fresh events should have been sent
+        assertEquals(2, drained)
+        assertNull(diskStore.read())
+
+        unmockkStatic(Log::class)
+    }
+
+    @Test
+    fun `concurrent drain calls do not duplicate sends`() = runTest {
+        mockkStatic(Log::class)
+        every { Log.d(any(), any()) } returns 0
+        every { Log.w(any(), any<String>()) } returns 0
+        every { Log.e(any(), any()) } returns 0
+        every { Log.e(any(), any(), any()) } returns 0
+
+
+        val overflowQueue = PersistableEventQueue(
+            maxCapacity = 5,
+            diskStore = diskStore,
+            maxOfflineDiskEvents = 100,
+            eventTTLMs = 0
+        )
+
+        val events = (1..10).map { createTestEvent("overflow-$it") }
+        diskStore.write(QueueSnapshot(version = 1, events = events))
+
+        val fakeClient = FakeNetworkClient()
+        fakeClient.nextResponse = NetworkResponse(200, emptyMap(), null)
+        val dispatcher = Dispatcher(
+            options = InitOptions(writeKey = "test-key", ingestionHost = "https://api.example.com", flushIntervalSeconds = 10),
+            queue = overflowQueue,
+            networkClient = fakeClient,
+            circuitBreaker = CircuitBreaker(),
+            scope = this,
+            config = DispatcherConfig()
+        )
+
+        // Launch two concurrent drain calls
+        val drain1 = async { overflowQueue.drainDiskOverflowToNetwork(dispatcher) }
+        val drain2 = async { overflowQueue.drainDiskOverflowToNetwork(dispatcher) }
+
+        val result1 = drain1.await()
+        val result2 = drain2.await()
+
+        // One should have drained all 10, the other should have returned 0 (guard)
+        val totalDrained = result1 + result2
+        assertEquals(10, totalDrained)
+        assertTrue("One drain should return 0", result1 == 0 || result2 == 0)
+        assertNull(diskStore.read())
+
+        unmockkStatic(Log::class)
+    }
+
+    @Test
+    fun `requeueToFront flushes memory queue to disk when at capacity`() {
+        val smallQueue = PersistableEventQueue(
+            maxCapacity = 5,
+            diskStore = diskStore,
+            maxOfflineDiskEvents = 100,
+            eventTTLMs = 0
+        )
+
+        // Simulate: memory queue has 3 new events (enqueued after a drain)
+        val newEvents = (1..3).map { createTestEvent("new-$it") }
+        newEvents.forEach { smallQueue.enqueue(it) }
+        assertEquals(3, smallQueue.size())
+
+        // Now requeue 5 older events (e.g., a failed send retry).
+        // Total would be 8, exceeding capacity of 5.
+        val requeued = (1..5).map { createTestEvent("older-$it") }
+        smallQueue.requeueToFront(requeued)
+
+        // Memory queue should have the 5 requeued events at the front (no drops)
+        assertEquals(5, smallQueue.size())
+        val drained = smallQueue.drain(10)
+        assertEquals(5, drained.size)
+        assertEquals("older-1", drained[0].messageId)
+        assertEquals("older-5", drained[4].messageId)
+
+        // The 3 new events should be on disk
+        val snapshot = diskStore.read()
+        assertNotNull(snapshot)
+        assertEquals(3, snapshot!!.events.size)
+        assertEquals("new-1", snapshot.events[0].messageId)
+        assertEquals("new-3", snapshot.events[2].messageId)
+    }
+
+
     private fun createTestEvent(messageId: String, timestamp: String = "2026-01-01T00:00:00.000Z"): EnrichedEventPayload {
         return EnrichedEventPayload(
             type = EventType.TRACK,