diff --git a/Assets/Plugins/StreamChat/Core/LowLevelClient/EventDrainStopwatch.cs b/Assets/Plugins/StreamChat/Core/LowLevelClient/EventDrainStopwatch.cs
new file mode 100644
index 00000000..062cf971
--- /dev/null
+++ b/Assets/Plugins/StreamChat/Core/LowLevelClient/EventDrainStopwatch.cs
@@ -0,0 +1,24 @@
+using System.Diagnostics;
+
+namespace StreamChat.Core.LowLevelClient
+{
+    /// <summary>
+    /// Elapsed-time source for the per-frame event drain budget. Production uses
+    /// <see cref="DiagnosticsElapsedStopwatch"/>; tests inject a fake so pacing is deterministic.
+    /// </summary>
+    internal interface IElapsedStopwatch
+    {
+        void Restart();
+
+        double ElapsedMilliseconds { get; }
+    }
+
+    internal sealed class DiagnosticsElapsedStopwatch : IElapsedStopwatch
+    {
+        public void Restart() => _stopwatch.Restart();
+
+        public double ElapsedMilliseconds => _stopwatch.Elapsed.TotalMilliseconds;
+
+        private readonly Stopwatch _stopwatch = new Stopwatch();
+    }
+}
diff --git a/Assets/Plugins/StreamChat/Core/LowLevelClient/EventDrainStopwatch.cs.meta b/Assets/Plugins/StreamChat/Core/LowLevelClient/EventDrainStopwatch.cs.meta
new file mode 100644
index 00000000..8b7e1da9
--- /dev/null
+++ b/Assets/Plugins/StreamChat/Core/LowLevelClient/EventDrainStopwatch.cs.meta
@@ -0,0 +1,3 @@
+fileFormatVersion: 2
+guid: e4d8a91c7b2f4e6a9d1c3b5f708294a6
+timeCreated: 1756152000
diff --git a/Assets/Plugins/StreamChat/Core/LowLevelClient/IStreamChatLowLevelClient.cs b/Assets/Plugins/StreamChat/Core/LowLevelClient/IStreamChatLowLevelClient.cs
index a9709bd4..9d1c035e 100644
--- a/Assets/Plugins/StreamChat/Core/LowLevelClient/IStreamChatLowLevelClient.cs
+++ b/Assets/Plugins/StreamChat/Core/LowLevelClient/IStreamChatLowLevelClient.cs
@@ -103,6 +103,11 @@ void SetReconnectStrategySettings(ReconnectStrategy reconnectStrategy, float? ex
         [Obsolete("Use DisconnectAsync(DisconnectCause). true maps to UserLogout, false to ConnectionReleased.")]
         Task DisconnectAsync(bool permanent);
 
+        /// <summary>
+        /// Fetch missed events via <c>/sync</c> and apply them. The returned task completes when
+        /// those events have been processed, which may span several <see cref="Update"/> calls
+        /// after a large catch-up. Keep calling <see cref="Update"/> while awaiting.
+        /// </summary>
         Task FetchAndProcessEventsSinceLastReceivedEvent(IEnumerable<string> channelCids);
 
         /// <summary>
diff --git a/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs b/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs
index d9d0d609..d03883fd 100644
--- a/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs
+++ b/Assets/Plugins/StreamChat/Core/LowLevelClient/StreamChatLowLevelClient.cs
@@ -26,7 +26,6 @@
 using StreamChat.Libs.Utils;
 using StreamChat.Libs.Websockets;
 using StreamChat.Core.LowLevelClient.Requests;
-using System.Linq;
 using StreamChat.Core.Helpers;
 using Thread = System.Threading.Thread;
 
@@ -218,6 +217,11 @@ private set
                 if (value == ConnectionState.Disconnected)
                 {
                     _disconnectionLastEventReceivedAt = _lastEventReceivedAt;
+                    ClearPendingHistoryEvents();
+
+                    // _heldLiveMessage is deliberately NOT cleared. IWebsocketClient's receive queue
+                    // survives the disconnect, so dropping just the one message we happened to hold
+                    // would punch a hole in an otherwise intact sequence.
                     RaiseDisconnected();
                 }
             }
@@ -430,13 +434,7 @@ public void Update(float deltaTime)
 
             _websocketClient.Update();
 
-            while (_websocketClient.TryDequeueMessage(out var msg))
-            {
-#if STREAM_DEBUG_ENABLED
-                _logs.Info(_authCredentials.UserId + " WS message: " + msg);
-#endif
-                HandleNewWebsocketMessage(msg, isLiveEvent: true);
-            }
+            DrainPendingEvents();
         }
 
         public bool IsLocalUser(User user) => user.Id == _authCredentials.UserId;
@@ -453,8 +451,20 @@ public void SetReconnectStrategySettings(ReconnectStrategy reconnectStrategy, fl
 
         public async Task FetchAndProcessEventsSinceLastReceivedEvent(IEnumerable<string> channelCids)
         {
+            var generation = _historyDrainGeneration;
             var response = await TrySyncHistoryAsync(channelCids);
-            ReplayHistoryEvents(response?.Events);
+
+            if (generation != _historyDrainGeneration)
+            {
+                return;
+            }
+
+            if (response?.Events == null || response.Events.Count == 0)
+            {
+                return;
+            }
+
+            await EnqueueHistoryEventsForReplay(response.Events);
         }
 
         /// <summary>
@@ -542,6 +552,8 @@ internal HistorySyncApplyResult ApplyHistoryEvents(IEnumerable<object> events)
                 return result;
             }
 
+            // Main-thread only. Must not interleave with the paced silent-batch drain:
+            // both share _isApplyingHistoryEvents and _historyMaxAppliedCreatedAt.
             _isApplyingHistoryEvents = true;
             _historyMaxAppliedCreatedAt = null;
 
@@ -603,6 +615,8 @@ private void TrackMaxAppliedCreatedAt(DateTimeOffset createdAt)
         public void Dispose()
         {
             ConnectionState = ConnectionState.Closing;
+            ClearPendingHistoryEvents();
+            _heldLiveMessage = null;
 
             _reconnectScheduler.Dispose();
 
@@ -635,6 +649,18 @@ public void Dispose()
 
         internal DisconnectCause LastDisconnectCause { get; private set; }
 
+        /// <summary>
+        /// Per-frame time budget for applying live WS + /sync history events. Tests may lower this.
+        /// </summary>
+        internal int EventDrainTimeBudgetMs { get; set; } = 3;
+
+        /// <summary>
+        /// Secondary bound so a batch of cheap events cannot still dump a huge list in one frame.
+        /// </summary>
+        internal int EventDrainCountCap { get; set; } = 64;
+
+        internal IElapsedStopwatch EventDrainStopwatch { get; set; } = new DiagnosticsElapsedStopwatch();
+
         internal void StopReconnectScheduler() => _reconnectScheduler.Stop();
 
         internal async Task<OwnUserInternalDTO> ConnectUserAsync(string apiKey, string userId,
@@ -683,6 +709,15 @@ internal async Task<OwnUserInternalDTO> ConnectUserAsync(string apiKey, string u
 
         private const string DefaultStreamAuthType = "jwt";
         private const int HealthCheckMaxWaitingTime = 30;
+        private const int EventDrainHealthGrace = 8;
+        private const int ReceiveBacklogWarningThreshold = 500;
+
+        /// <summary>
+        /// How long a paced /sync drain may block live <c>health.check</c> before the receive-side
+        /// timeout is allowed to fire. Shorter reintroduces a reconnect loop; longer delays
+        /// detecting a genuinely dead socket.
+        /// </summary>
+        internal const float EventDrainMaxStallSeconds = 120f;
 
         // For WebGL there is a slight delay when sending therefore we send HC event a bit sooner just in case
         private const int HealthCheckSendInterval = HealthCheckMaxWaitingTime - 1;
@@ -704,6 +739,8 @@ internal async Task<OwnUserInternalDTO> ConnectUserAsync(string apiKey, string u
 
         private readonly object _websocketConnectionFailedFlagLock = new object();
         private readonly object _websocketDisconnectedFlagLock = new object();
+        private readonly object _pendingHistoryEventsLock = new object();
+        private readonly Queue<PendingHistoryEvent> _pendingHistoryEvents = new Queue<PendingHistoryEvent>();
 
         /// <summary>
         /// Every <see cref="ConnectionState"/> write must happen on this thread
@@ -737,8 +774,13 @@ internal async Task<OwnUserInternalDTO> ConnectUserAsync(string apiKey, string u
         /// </summary>
         private DateTimeOffset? _disconnectionLastEventReceivedAt;
 
+        private string _heldLiveMessage;
+        private TaskCompletionSource<bool> _historyDrainTcs;
+        private int _historyDrainGeneration;
         private bool _isApplyingHistoryEvents;
         private DateTimeOffset? _historyMaxAppliedCreatedAt;
+        private float? _historyDrainStartedAt;
+        private bool _receiveBacklogWarningArmed = true;
 
         private async Task RefreshAuthTokenFromProvider()
         {
@@ -1113,8 +1155,7 @@ private void RegisterEventType<TDto, TEvent>(string key,
                 try
                 {
 #if STREAM_DEBUG_ENABLED
-                    var ignoreKeys = new[] { WSEventType.HealthCheck };
-                    if (!ignoreKeys.Contains(key))
+                    if (key != WSEventType.HealthCheck)
                     {
                         _logs.Warning("WS event received KEY: " + key + " CONTENT: " + payload);
                     }
@@ -1162,6 +1203,291 @@ private TEvent DeserializeEvent<TDto, TEvent>(object payload, out TDto dto)
             return response;
         }
 
+        private void DrainPendingEvents()
+        {
+            EventDrainStopwatch.Restart();
+            var processed = 0;
+            var countCap = EventDrainCountCap;
+            var timeBudgetMs = EventDrainTimeBudgetMs;
+
+            bool BudgetExhausted()
+                => processed > 0 && (processed >= countCap ||
+                                     EventDrainStopwatch.ElapsedMilliseconds >= timeBudgetMs);
+
+            while (true)
+            {
+                if (HasPendingHistoryEvents())
+                {
+                    if (BudgetExhausted())
+                    {
+                        break;
+                    }
+
+                    if (!TryDequeueHistoryEvent(out var historyEvent))
+                    {
+                        continue;
+                    }
+
+                    ProcessHistoryEvent(historyEvent);
+                    processed++;
+                    TryCompleteHistoryDrainIfIdle();
+                    continue;
+                }
+
+                if (!TryTakeLiveMessage(out var liveMsg))
+                {
+                    break;
+                }
+
+#if STREAM_DEBUG_ENABLED
+                _logs.Info(_authCredentials.UserId + " WS message: " + liveMsg);
+#endif
+
+                var hardStop = processed >= countCap + EventDrainHealthGrace;
+
+                // IsHealthCheckMessage does a full JObject.Parse, so only pay for it when the
+                // budget is actually spent and we need to decide whether to make an exception.
+                if (hardStop || (BudgetExhausted() && !IsHealthCheckMessage(liveMsg)))
+                {
+                    _heldLiveMessage = liveMsg;
+                    break;
+                }
+
+                HandleNewWebsocketMessage(liveMsg, isLiveEvent: true);
+                processed++;
+            }
+
+            TryCompleteHistoryDrainIfIdle();
+            WarnOnStalledPump();
+        }
+
+        private bool TryTakeLiveMessage(out string msg)
+        {
+            if (_heldLiveMessage != null)
+            {
+                msg = _heldLiveMessage;
+                _heldLiveMessage = null;
+                return true;
+            }
+
+            return _websocketClient.TryDequeueMessage(out msg);
+        }
+
+        private bool IsHealthCheckMessage(string msg)
+        {
+            if (string.IsNullOrEmpty(msg))
+            {
+                return false;
+            }
+
+            try
+            {
+                return _serializer.TryPeekValue<string>(msg, "type", out var type)
+                       && type == WSEventType.HealthCheck;
+            }
+            catch (Exception)
+            {
+                return false;
+            }
+        }
+
+        internal Task EnqueueHistoryEvents(IEnumerable<object> events)
+            => EnqueueHistoryEventsForReplay(events);
+
+        /// <summary>
+        /// Queue /sync history events to be replayed as normal events across several Update calls.
+        /// The returned task completes when the last one has been handled.
+        /// </summary>
+        internal Task EnqueueHistoryEventsForReplay(IEnumerable<object> events)
+            => EnqueueHistoryEvents(events, silentBatch: false);
+
+        /// <summary>
+        /// Queue /sync history events to be applied to local state without per-event public callbacks,
+        /// across several Update calls. The /sync watermark advances once, when the whole batch drains.
+        /// </summary>
+        internal Task EnqueueHistoryEventsForBatchApply(IEnumerable<object> events)
+            => EnqueueHistoryEvents(events, silentBatch: true);
+
+        private readonly struct PendingHistoryEvent
+        {
+            public PendingHistoryEvent(object payload, bool silentBatch)
+            {
+                Payload = payload;
+                SilentBatch = silentBatch;
+            }
+
+            public readonly object Payload;
+            public readonly bool SilentBatch;
+        }
+
+        private Task EnqueueHistoryEvents(IEnumerable<object> events, bool silentBatch)
+        {
+            if (events == null)
+            {
+                return Task.CompletedTask;
+            }
+
+            lock (_pendingHistoryEventsLock)
+            {
+                foreach (var e in events)
+                {
+                    _pendingHistoryEvents.Enqueue(new PendingHistoryEvent(e, silentBatch));
+                }
+
+                if (_pendingHistoryEvents.Count == 0)
+                {
+                    return Task.CompletedTask;
+                }
+
+                if (_historyDrainTcs == null || _historyDrainTcs.Task.IsCompleted)
+                {
+                    // TrySetResult is called outside this lock. Do not add
+                    // RunContinuationsAsynchronously: RestoreStateLostDuringDisconnect awaits this
+                    // task and must continue on the same Update so the re-query starts only after
+                    // history has been applied. An inline continuation cannot deadlock because we
+                    // already released the lock.
+                    _historyDrainTcs = new TaskCompletionSource<bool>();
+                }
+
+                return _historyDrainTcs.Task;
+            }
+        }
+
+        private bool HasPendingHistoryEvents()
+        {
+            lock (_pendingHistoryEventsLock)
+            {
+                return _pendingHistoryEvents.Count > 0;
+            }
+        }
+
+        private bool TryDequeueHistoryEvent(out PendingHistoryEvent historyEvent)
+        {
+            lock (_pendingHistoryEventsLock)
+            {
+                if (_pendingHistoryEvents.Count == 0)
+                {
+                    historyEvent = default;
+                    return false;
+                }
+
+                historyEvent = _pendingHistoryEvents.Dequeue();
+                return true;
+            }
+        }
+
+        private void ProcessHistoryEvent(PendingHistoryEvent pending)
+        {
+            if (pending.Payload == null)
+            {
+                return;
+            }
+
+            if (!pending.SilentBatch)
+            {
+                try
+                {
+                    HandleNewWebsocketMessage(pending.Payload);
+                }
+                catch (Exception e)
+                {
+                    // A malformed /sync event must not throw out of Update.
+                    _logs.Exception(e);
+                }
+
+                return;
+            }
+
+            _isApplyingHistoryEvents = true;
+            try
+            {
+                HandleNewWebsocketMessage(pending.Payload);
+            }
+            catch (Exception e)
+            {
+                // Do not abort the batch. The /sync last_sync_at cursor only moves to events that
+                // applied, so a failed event is retried on the next reconnect.
+                _logs.Exception(e);
+            }
+            finally
+            {
+                _isApplyingHistoryEvents = false;
+            }
+        }
+
+        private void TryCompleteHistoryDrainIfIdle()
+        {
+            TaskCompletionSource<bool> tcs;
+            lock (_pendingHistoryEventsLock)
+            {
+                if (_pendingHistoryEvents.Count > 0 || _historyDrainTcs == null)
+                {
+                    return;
+                }
+
+                tcs = _historyDrainTcs;
+                _historyDrainTcs = null;
+            }
+
+            // End of a silent batch: advance the /sync cursor exactly once, to the newest event
+            // that actually applied. Mirrors the finally block in ApplyHistoryEvents.
+            if (_historyMaxAppliedCreatedAt.HasValue)
+            {
+                TryAdvanceLastEventReceivedAt(_historyMaxAppliedCreatedAt.Value, HistorySyncWatermarkSource);
+            }
+
+            _historyMaxAppliedCreatedAt = null;
+
+            // Outside the lock so FetchAndProcess / recovery resumes on this Update, not a later
+            // scheduler tick.
+            tcs.TrySetResult(true);
+        }
+
+        private void ClearPendingHistoryEvents()
+        {
+            TaskCompletionSource<bool> tcs;
+            lock (_pendingHistoryEventsLock)
+            {
+                _historyDrainGeneration++;
+                _pendingHistoryEvents.Clear();
+                tcs = _historyDrainTcs;
+                _historyDrainTcs = null;
+            }
+
+            // A partially applied, abandoned batch must not later advance the cursor past events
+            // that were dropped.
+            _historyMaxAppliedCreatedAt = null;
+            _historyDrainStartedAt = null;
+
+            tcs?.TrySetResult(true);
+        }
+
+        private void WarnOnStalledPump()
+        {
+            int pendingHistory;
+            lock (_pendingHistoryEventsLock)
+            {
+                pendingHistory = _pendingHistoryEvents.Count;
+            }
+
+            var backlog = pendingHistory + (_heldLiveMessage != null ? 1 : 0);
+            if (backlog > ReceiveBacklogWarningThreshold)
+            {
+                if (_receiveBacklogWarningArmed)
+                {
+                    _receiveBacklogWarningArmed = false;
+                    _logs.Warning(
+                        $"{pendingHistory} chat events are still waiting to be applied. They are being paced across " +
+                        $"frames ({EventDrainTimeBudgetMs}ms per Update). Nothing is dropped, but your event handlers " +
+                        "may be too expensive, or Update is not being called often enough.");
+                }
+
+                return;
+            }
+
+            _receiveBacklogWarningArmed = true;
+        }
+
         private TDto DeserializePayload<TDto>(object payload)
         {
             if (payload is string content)
@@ -1387,6 +1713,39 @@ private void UpdateHealthCheck()
                 PingHealthCheck();
             }
 
+            // A paced history drain blocks live messages, including health.check, so the receive
+            // timestamp cannot advance while it runs. Timing out here would clear the queue and
+            // make the next /sync re-fetch the same events - a loop that never catches up. Pings
+            // still go out above, so the server keeps the socket open. Bounded by
+            // EventDrainMaxStallSeconds so a genuinely dead socket is still detected.
+            if (HasPendingHistoryEvents())
+            {
+                if (!_historyDrainStartedAt.HasValue)
+                {
+                    _historyDrainStartedAt = _timeService.Time;
+                }
+
+                if (_timeService.Time - _historyDrainStartedAt.Value < EventDrainMaxStallSeconds)
+                {
+                    _lastHealthCheckReceivedTime = _timeService.Time;
+                    return;
+                }
+
+                _logs.Warning(
+                    $"The /sync catch-up has been draining for more than {EventDrainMaxStallSeconds}s and is " +
+                    "still behind. Letting the health-check timeout apply. Your event handlers are likely too " +
+                    "expensive for the per-frame budget.");
+
+                // We have been stamping _lastHealthCheckReceivedTime every frame to hold the
+                // timeout off. Expire it so the check below fires this frame instead of waiting
+                // another HealthCheckMaxWaitingTime after we stop stamping.
+                _lastHealthCheckReceivedTime = _timeService.Time - HealthCheckMaxWaitingTime - 1f;
+            }
+            else
+            {
+                _historyDrainStartedAt = null;
+            }
+
             var timeSinceLastHealthCheck = _timeService.Time - _lastHealthCheckReceivedTime;
             if (timeSinceLastHealthCheck > HealthCheckMaxWaitingTime)
             {
diff --git a/Assets/Plugins/StreamChat/Core/StreamChatClient.cs b/Assets/Plugins/StreamChat/Core/StreamChatClient.cs
index 90eee900..769dab48 100644
--- a/Assets/Plugins/StreamChat/Core/StreamChatClient.cs
+++ b/Assets/Plugins/StreamChat/Core/StreamChatClient.cs
@@ -1237,7 +1237,9 @@ private void OnConnected(HealthCheckEventInternalDTO dto)
         /// in local state but stop receiving events.
         ///
         /// 1. <c>/sync</c> catch-up (best effort). Must run first: a replayed <c>channel.truncated</c>
-        ///    would wipe messages fetched in step 2.
+        ///    would wipe messages fetched in step 2. A large catch-up is paced across Update
+        ///    calls; this method awaits that drain, so step 2 does not start until it finishes.
+        ///    <see cref="StateRecovered"/> may therefore fire seconds after <see cref="Connected"/>.
         /// 2. Re-query and re-watch, even if step 1 failed or was skipped.
         /// 3. Raise <see cref="StateRecovered"/> once.
         ///
@@ -1384,11 +1386,11 @@ private async Task TryCatchUpWithHistoryAsync(IReadOnlyList<string> recoveryChan
 
                 if (InternalLowLevelClient.Config.StateRecoveryStrategy == StateRecoveryStrategy.BatchStateUpdate)
                 {
-                    InternalLowLevelClient.ApplyHistoryEvents(response.Events);
+                    await InternalLowLevelClient.EnqueueHistoryEventsForBatchApply(response.Events);
                 }
                 else
                 {
-                    InternalLowLevelClient.ReplayHistoryEvents(response.Events);
+                    await InternalLowLevelClient.EnqueueHistoryEventsForReplay(response.Events);
                 }
             }
             catch (StreamApiException ex) when (ex.IsInputError())
diff --git a/Assets/Plugins/StreamChat/Tests/LowLevelClient/StreamChatLowLevelClientEventDrainTests.cs b/Assets/Plugins/StreamChat/Tests/LowLevelClient/StreamChatLowLevelClientEventDrainTests.cs
new file mode 100644
index 00000000..b99b3778
--- /dev/null
+++ b/Assets/Plugins/StreamChat/Tests/LowLevelClient/StreamChatLowLevelClientEventDrainTests.cs
@@ -0,0 +1,440 @@
+#if STREAM_TESTS_ENABLED
+using System;
+using System.Collections.Generic;
+using System.Net.WebSockets;
+using System.Reflection;
+using System.Threading.Tasks;
+using NSubstitute;
+using NUnit.Framework;
+using StreamChat.Core;
+using StreamChat.Core.Configs;
+using StreamChat.Core.LowLevelClient;
+using StreamChat.Core.LowLevelClient.Events;
+using StreamChat.Libs.AppInfo;
+using StreamChat.Libs.Auth;
+using StreamChat.Libs.Http;
+using StreamChat.Libs.Logs;
+using StreamChat.Libs.NetworkMonitors;
+using StreamChat.Libs.Serialization;
+using StreamChat.Libs.Time;
+using StreamChat.Libs.Websockets;
+
+namespace StreamChat.Tests.LowLevelClient
+{
+    internal class StreamChatLowLevelClientEventDrainTests
+    {
+        [SetUp]
+        public void Up()
+        {
+            _authCredentials = new AuthCredentials("api123", "token123", "user123");
+            _mockWebsocketClient = Substitute.For<IWebsocketClient>();
+            _mockHttpClient = Substitute.For<IHttpClient>();
+            _mockTimeService = Substitute.For<ITimeService>();
+            _mockNetworkMonitor = Substitute.For<INetworkMonitor>();
+            _mockApplicationInfo = Substitute.For<IApplicationInfo>();
+            _mockLogs = new UnityLogs();
+            _mockStreamClientConfig = Substitute.For<IStreamClientConfig>();
+
+            _mockWebsocketClient.ConnectAsync(Arg.Any<Uri>()).Returns(Task.CompletedTask);
+            _mockHttpClient
+                .SendHttpRequestAsync(Arg.Is(HttpMethodType.Post), Arg.Any<Uri>(), Arg.Any<object>())
+                .Returns(new HttpResponse(true, 200, "{\"events\":[]}", null, null));
+        }
+
+        [TearDown]
+        public void TearDown()
+        {
+            for (int i = _resourcesToDispose.Count - 1; i >= 0; i--)
+            {
+                _resourcesToDispose[i].Dispose();
+            }
+
+            _resourcesToDispose.Clear();
+        }
+
+        [Test]
+        public void when_live_backlog_exceeds_count_cap_expect_one_update_handles_only_the_cap()
+        {
+            var client = CreateConnectedClient();
+            client.EventDrainCountCap = 2;
+            client.EventDrainTimeBudgetMs = 10_000;
+
+            var received = new List<string>();
+            client.MessageReceived += e => received.Add(e.Message.Id);
+
+            EnqueueLiveMessages(Message("m1"), Message("m2"), Message("m3"), Message("m4"), Message("m5"));
+            client.Update(0.1f);
+
+            Assert.AreEqual(new[] { "m1", "m2" }, received.ToArray());
+
+            client.Update(0.1f);
+            Assert.AreEqual(new[] { "m1", "m2", "m3", "m4" }, received.ToArray());
+
+            client.Update(0.1f);
+            Assert.AreEqual(new[] { "m1", "m2", "m3", "m4", "m5" }, received.ToArray());
+        }
+
+        [Test]
+        public void when_one_live_message_per_update_expect_it_drains_fully()
+        {
+            var client = CreateConnectedClient();
+            client.EventDrainCountCap = 2;
+
+            var received = 0;
+            client.MessageReceived += _ => received++;
+
+            EnqueueLiveMessages(Message("m1"));
+            client.Update(0.1f);
+
+            Assert.AreEqual(1, received);
+
+            client.Update(0.1f);
+            Assert.AreEqual(1, received);
+        }
+
+        [Test]
+        public void when_budget_spent_expect_live_health_check_still_handled()
+        {
+            var client = CreateConnectedClient();
+            client.EventDrainCountCap = 1;
+            client.EventDrainTimeBudgetMs = 10_000;
+
+            var messages = new List<string>();
+            var healthChecks = 0;
+            client.MessageReceived += e => messages.Add(e.Message.Id);
+            client.EventReceived += payload =>
+            {
+                if (payload.Contains("health.check"))
+                {
+                    healthChecks++;
+                }
+            };
+
+            EnqueueLiveMessages(Message("m1"), HealthCheck(), Message("m2"));
+            client.Update(0.1f);
+
+            Assert.AreEqual(new[] { "m1" }, messages.ToArray());
+            Assert.GreaterOrEqual(healthChecks, 1);
+
+            client.Update(0.1f);
+            Assert.AreEqual(new[] { "m1", "m2" }, messages.ToArray());
+        }
+
+        [Test]
+        public void when_time_budget_exhausted_expect_remaining_live_events_deferred()
+        {
+            var client = CreateConnectedClient();
+            var clock = new ManualElapsedStopwatch();
+            client.EventDrainStopwatch = clock;
+            client.EventDrainTimeBudgetMs = 3;
+            client.EventDrainCountCap = 1000;
+
+            var received = new List<string>();
+            client.MessageReceived += e =>
+            {
+                received.Add(e.Message.Id);
+                clock.ElapsedMilliseconds = 10;
+            };
+
+            EnqueueLiveMessages(Message("m1"), Message("m2"), Message("m3"));
+            client.Update(0.1f);
+
+            Assert.AreEqual(new[] { "m1" }, received.ToArray());
+
+            clock.ElapsedMilliseconds = 0;
+            client.Update(0.1f);
+            Assert.AreEqual(new[] { "m1", "m2" }, received.ToArray());
+        }
+
+        [Test]
+        public void when_history_and_live_queued_expect_history_processed_first()
+        {
+            var client = CreateConnectedClient();
+            client.EventDrainCountCap = 10;
+            client.EventDrainTimeBudgetMs = 10_000;
+
+            var received = new List<string>();
+            client.MessageReceived += e => received.Add(e.Message.Id);
+
+            StubSyncEvents(MessageEventJson("h1"), MessageEventJson("h2"));
+            SetDisconnectionWatermark(client, new DateTimeOffset(2026, 8, 10, 11, 0, 0, TimeSpan.Zero));
+            _mockTimeService.Now.Returns(new DateTimeOffset(2026, 8, 10, 12, 0, 0, TimeSpan.Zero));
+
+            EnqueueLiveMessages(Message("l1"), Message("l2"));
+
+            var fetch = client.FetchAndProcessEventsSinceLastReceivedEvent(new[] { "messaging:test" });
+            client.Update(0.1f);
+
+            Assert.AreEqual(new[] { "h1", "h2", "l1", "l2" }, received.ToArray());
+            Assert.IsTrue(fetch.IsCompleted);
+            fetch.GetAwaiter().GetResult();
+        }
+
+        [Test]
+        public void when_sync_returns_many_events_expect_they_span_updates_and_task_waits()
+        {
+            var client = CreateConnectedClient();
+            client.EventDrainCountCap = 2;
+            client.EventDrainTimeBudgetMs = 10_000;
+
+            var received = new List<string>();
+            client.MessageReceived += e => received.Add(e.Message.Id);
+
+            StubSyncEvents(MessageEventJson("h1"), MessageEventJson("h2"), MessageEventJson("h3"),
+                MessageEventJson("h4"), MessageEventJson("h5"));
+            SetDisconnectionWatermark(client, new DateTimeOffset(2026, 8, 10, 11, 0, 0, TimeSpan.Zero));
+            _mockTimeService.Now.Returns(new DateTimeOffset(2026, 8, 10, 12, 0, 0, TimeSpan.Zero));
+
+            var fetch = client.FetchAndProcessEventsSinceLastReceivedEvent(new[] { "messaging:test" });
+            Assert.IsFalse(fetch.IsCompleted);
+
+            client.Update(0.1f);
+            Assert.AreEqual(new[] { "h1", "h2" }, received.ToArray());
+            Assert.IsFalse(fetch.IsCompleted);
+
+            client.Update(0.1f);
+            Assert.AreEqual(new[] { "h1", "h2", "h3", "h4" }, received.ToArray());
+
+            client.Update(0.1f);
+            Assert.AreEqual(new[] { "h1", "h2", "h3", "h4", "h5" }, received.ToArray());
+            Assert.IsTrue(fetch.IsCompleted);
+            fetch.GetAwaiter().GetResult();
+        }
+
+        [Test]
+        public void when_disconnect_mid_history_drain_expect_pending_cleared_and_no_double_apply()
+        {
+            var client = CreateConnectedClient();
+            client.EventDrainCountCap = 1;
+            client.EventDrainTimeBudgetMs = 10_000;
+            SetupDisconnectRaisesDisconnected();
+
+            var received = new List<string>();
+            client.MessageReceived += e => received.Add(e.Message.Id);
+
+            StubSyncEvents(MessageEventJson("h1"), MessageEventJson("h2"), MessageEventJson("h3"));
+            SetDisconnectionWatermark(client, new DateTimeOffset(2026, 8, 10, 11, 0, 0, TimeSpan.Zero));
+            _mockTimeService.Now.Returns(new DateTimeOffset(2026, 8, 10, 12, 0, 0, TimeSpan.Zero));
+
+            var fetch = client.FetchAndProcessEventsSinceLastReceivedEvent(new[] { "messaging:test" });
+            client.Update(0.1f);
+            Assert.AreEqual(new[] { "h1" }, received.ToArray());
+
+            client.DisconnectAsync(DisconnectCause.ConnectionReleased).GetAwaiter().GetResult();
+            Assert.IsTrue(fetch.IsCompleted);
+
+            client.Update(0.1f);
+            Assert.AreEqual(new[] { "h1" }, received.ToArray());
+        }
+
+        [Test]
+        public void when_health_timeout_elapses_during_history_drain_expect_connection_stays()
+        {
+            var client = CreateConnectedClient();
+            client.EventDrainCountCap = 1;
+            client.EventDrainTimeBudgetMs = 10_000;
+            SetupDisconnectRaisesDisconnected();
+
+            EnqueueHistory(client, Message("h1"), Message("h2"), Message("h3"), Message("h4"), Message("h5"));
+            client.Update(0.1f);
+            Assert.AreEqual(ConnectionState.Connected, client.ConnectionState);
+
+            _mockTimeService.Time.Returns(31f);
+            client.Update(0.1f);
+
+            Assert.AreEqual(ConnectionState.Connected, client.ConnectionState);
+        }
+
+        [Test]
+        public void when_history_drain_exceeds_max_stall_expect_health_timeout_fires()
+        {
+            var client = CreateConnectedClient();
+            client.EventDrainCountCap = 1;
+            client.EventDrainTimeBudgetMs = 10_000;
+            SetupDisconnectRaisesDisconnected();
+
+            EnqueueHistory(client, Message("h1"), Message("h2"), Message("h3"), Message("h4"), Message("h5"));
+            client.Update(0.1f);
+
+            _mockTimeService.Time.Returns(StreamChatLowLevelClient.EventDrainMaxStallSeconds + 1f);
+            client.Update(0.1f);
+
+            Assert.AreNotEqual(ConnectionState.Connected, client.ConnectionState);
+        }
+
+        [Test]
+        public void when_disconnect_with_held_live_message_expect_it_is_still_delivered()
+        {
+            var client = CreateConnectedClient();
+            client.EventDrainCountCap = 1;
+            client.EventDrainTimeBudgetMs = 10_000;
+            SetupDisconnectRaisesDisconnected();
+
+            var received = new List<string>();
+            client.MessageReceived += e => received.Add(e.Message.Id);
+
+            EnqueueLiveMessages(Message("m1"), Message("m2"));
+            client.Update(0.1f);
+            Assert.AreEqual(new[] { "m1" }, received.ToArray());
+
+            client.DisconnectAsync(DisconnectCause.ConnectionReleased).GetAwaiter().GetResult();
+            client.Update(0.1f);
+
+            Assert.AreEqual(new[] { "m1", "m2" }, received.ToArray());
+        }
+
+        [Test]
+        public void when_silent_batch_spans_updates_expect_watermark_advances_once_at_end()
+        {
+            var client = CreateConnectedClient();
+            client.EventDrainCountCap = 1;
+            client.EventDrainTimeBudgetMs = 10_000;
+
+            var t0 = new DateTimeOffset(2026, 8, 10, 12, 0, 0, TimeSpan.Zero);
+            var t1 = t0.AddMinutes(1);
+            var t2 = t0.AddMinutes(2);
+
+            var received = 0;
+            client.MessageReceived += _ => received++;
+
+            var drain = client.EnqueueHistoryEventsForBatchApply(new object[]
+            {
+                Message("m1", t0),
+                Message("m2", t2),
+                Message("m3", t1),
+            });
+
+            client.Update(0.1f);
+            Assert.AreEqual(0, received);
+            Assert.IsNull(GetLastEventReceivedAt(client));
+            Assert.IsFalse(drain.IsCompleted);
+
+            client.Update(0.1f);
+            Assert.AreEqual(0, received);
+            Assert.IsNull(GetLastEventReceivedAt(client));
+            Assert.IsFalse(drain.IsCompleted);
+
+            client.Update(0.1f);
+            Assert.AreEqual(0, received);
+            Assert.AreEqual(t2, GetLastEventReceivedAt(client));
+            Assert.IsTrue(drain.IsCompleted);
+        }
+
+        [Test]
+        public void when_held_live_then_history_enqueued_expect_history_before_held()
+        {
+            var client = CreateConnectedClient();
+            client.EventDrainCountCap = 1;
+            client.EventDrainTimeBudgetMs = 10_000;
+
+            var received = new List<string>();
+            client.MessageReceived += e => received.Add(e.Message.Id);
+
+            EnqueueLiveMessages(Message("l1"), Message("l2"));
+            client.Update(0.1f);
+            Assert.AreEqual(new[] { "l1" }, received.ToArray());
+
+            EnqueueHistory(client, Message("h1"));
+            client.Update(0.1f);
+
+            Assert.AreEqual(new[] { "l1", "h1" }, received.ToArray());
+
+            client.Update(0.1f);
+            Assert.AreEqual(new[] { "l1", "h1", "l2" }, received.ToArray());
+        }
+
+        private readonly List<IDisposable> _resourcesToDispose = new List<IDisposable>();
+
+        private AuthCredentials _authCredentials;
+        private IWebsocketClient _mockWebsocketClient;
+        private IApplicationInfo _mockApplicationInfo;
+        private ILogs _mockLogs;
+        private ITimeService _mockTimeService;
+        private INetworkMonitor _mockNetworkMonitor;
+        private IHttpClient _mockHttpClient;
+        private IStreamClientConfig _mockStreamClientConfig;
+
+        private StreamChatLowLevelClient CreateConnectedClient()
+        {
+            var client = new StreamChatLowLevelClient(_authCredentials, _mockWebsocketClient, _mockHttpClient,
+                new NewtonsoftJsonSerializer(), _mockTimeService, _mockNetworkMonitor, _mockApplicationInfo,
+                _mockLogs, _mockStreamClientConfig);
+            _resourcesToDispose.Add(client);
+
+            EnqueueLiveMessages(HealthCheck());
+            client.Connect();
+            client.Update(0.2f);
+            Assert.AreEqual(ConnectionState.Connected, client.ConnectionState);
+            return client;
+        }
+
+        private void EnqueueHistory(StreamChatLowLevelClient client, params object[] events)
+        {
+            client.EnqueueHistoryEventsForReplay(events);
+        }
+
+        private void EnqueueLiveMessages(params string[] messages)
+        {
+            var queue = new Queue<string>(messages);
+            _mockWebsocketClient.TryDequeueMessage(out Arg.Any<string>()).Returns(arg =>
+            {
+                if (queue.Count == 0)
+                {
+                    return false;
+                }
+
+                arg[0] = queue.Dequeue();
+                return true;
+            });
+        }
+
+        private void StubSyncEvents(params string[] eventJsonObjects)
+        {
+            var body = "{\"events\":[" + string.Join(",", eventJsonObjects) + "]}";
+            _mockHttpClient
+                .SendHttpRequestAsync(Arg.Is(HttpMethodType.Post), Arg.Any<Uri>(), Arg.Any<object>())
+                .Returns(new HttpResponse(true, 200, body, null, null));
+        }
+
+        private void SetupDisconnectRaisesDisconnected()
+        {
+            _mockWebsocketClient.When(_ => _.DisconnectAsync(Arg.Any<WebSocketCloseStatus>(), Arg.Any<string>()))
+                .Do(_ => { _mockWebsocketClient.Disconnected += Raise.Event<Action>(); });
+        }
+
+        private static void SetDisconnectionWatermark(StreamChatLowLevelClient client, DateTimeOffset value)
+        {
+            var field = typeof(StreamChatLowLevelClient).GetField("_disconnectionLastEventReceivedAt",
+                BindingFlags.Instance | BindingFlags.NonPublic);
+            Assert.IsNotNull(field);
+            field.SetValue(client, (DateTimeOffset?)value);
+        }
+
+        private static DateTimeOffset? GetLastEventReceivedAt(StreamChatLowLevelClient client)
+        {
+            var field = typeof(StreamChatLowLevelClient).GetField("_lastEventReceivedAt",
+                BindingFlags.Instance | BindingFlags.NonPublic);
+            Assert.IsNotNull(field);
+            return (DateTimeOffset?)field.GetValue(client);
+        }
+
+        private static string Message(string id)
+            => $"{{\"type\":\"message.new\",\"cid\":\"messaging:test\",\"message\":{{\"id\":\"{id}\",\"text\":\"{id}\",\"type\":\"regular\"}}}}";
+
+        private static string Message(string id, DateTimeOffset createdAt)
+            => $"{{\"type\":\"message.new\",\"cid\":\"messaging:test\",\"created_at\":\"{createdAt:O}\"," +
+               $"\"message\":{{\"id\":\"{id}\",\"text\":\"{id}\",\"type\":\"regular\",\"created_at\":\"{createdAt:O}\"}}}}";
+
+        private static string MessageEventJson(string id) => Message(id);
+
+        private static string HealthCheck() => "{\"connection_id\":\"fakeId\",\"type\":\"health.check\"}";
+
+        private sealed class ManualElapsedStopwatch : IElapsedStopwatch
+        {
+            public double ElapsedMilliseconds { get; set; }
+
+            public void Restart() => ElapsedMilliseconds = 0;
+        }
+    }
+}
+#endif
diff --git a/Assets/Plugins/StreamChat/Tests/LowLevelClient/StreamChatLowLevelClientEventDrainTests.cs.meta b/Assets/Plugins/StreamChat/Tests/LowLevelClient/StreamChatLowLevelClientEventDrainTests.cs.meta
new file mode 100644
index 00000000..9871d70e
--- /dev/null
+++ b/Assets/Plugins/StreamChat/Tests/LowLevelClient/StreamChatLowLevelClientEventDrainTests.cs.meta
@@ -0,0 +1,3 @@
+fileFormatVersion: 2
+guid: b7c2e19f4a8d4c3e9f1b6a5d2c8e0f34
+timeCreated: 1756152100
diff --git a/Assets/Plugins/StreamChat/Tests/StateSync/StateRecoveryClientTests.cs b/Assets/Plugins/StreamChat/Tests/StateSync/StateRecoveryClientTests.cs
index 6f786a59..c84d38c3 100644
--- a/Assets/Plugins/StreamChat/Tests/StateSync/StateRecoveryClientTests.cs
+++ b/Assets/Plugins/StreamChat/Tests/StateSync/StateRecoveryClientTests.cs
@@ -549,6 +549,95 @@ public void when_replay_recovery_syncs_message_new_expect_channel_message_receiv
             Assert.AreEqual(1, _recoveredEvents.Count);
         }
 
+        [Test]
+        public void when_replay_recovery_syncs_many_events_expect_they_span_updates()
+        {
+            _client.InternalLowLevelClient.EventDrainCountCap = 2;
+            _client.InternalLowLevelClient.EventDrainTimeBudgetMs = 10_000;
+
+            Connect();
+            var channel = WatchChannel("messaging:a");
+            RespondWith(QueryChannelsEndpoint, QueryChannelsJson("messaging:a", "a"));
+
+            var received = 0;
+            channel.MessageReceived += (_, __) => received++;
+
+            ReconnectWithSync(ManySyncMessages("messaging:a", count: 5));
+
+            Assert.AreEqual(1, received, "Health check consumes one slot of the cap; only one history event fits.");
+            AssertQueryChannelsCallCount(0);
+            Assert.AreEqual(0, _recoveredEvents.Count);
+
+            Update();
+            Assert.AreEqual(3, received);
+            AssertQueryChannelsCallCount(0);
+            Assert.AreEqual(0, _recoveredEvents.Count);
+
+            Update();
+            Assert.AreEqual(5, received);
+            AssertQueryChannelsCallCount(1);
+            Assert.AreEqual(1, _recoveredEvents.Count);
+        }
+
+        [Test]
+        public void when_silent_recovery_syncs_many_events_expect_they_span_updates_without_callbacks()
+        {
+            _config.StateRecoveryStrategy = StateRecoveryStrategy.BatchStateUpdate;
+            _client.InternalLowLevelClient.EventDrainCountCap = 2;
+            _client.InternalLowLevelClient.EventDrainTimeBudgetMs = 10_000;
+
+            Connect();
+            var channel = WatchChannel("messaging:a");
+            RespondWith(QueryChannelsEndpoint, QueryChannelsJson("messaging:a", "a"));
+
+            var received = 0;
+            channel.MessageReceived += (_, __) => received++;
+
+            ReconnectWithSync(ManySyncMessages("messaging:a", count: 5));
+
+            Assert.AreEqual(0, received);
+            Assert.AreEqual(1, channel.Messages.Count);
+            AssertQueryChannelsCallCount(0);
+            Assert.AreEqual(0, _recoveredEvents.Count);
+            Assert.IsNull(GetLastEventReceivedAt(_client.InternalLowLevelClient));
+
+            Update();
+            Assert.AreEqual(0, received);
+            Assert.AreEqual(3, channel.Messages.Count);
+            AssertQueryChannelsCallCount(0);
+            Assert.IsNull(GetLastEventReceivedAt(_client.InternalLowLevelClient));
+
+            Update();
+            Assert.AreEqual(0, received);
+            Assert.AreEqual(5, channel.Messages.Count);
+            AssertQueryChannelsCallCount(1);
+            Assert.AreEqual(1, _recoveredEvents.Count);
+            Assert.AreEqual(new DateTimeOffset(2026, 8, 24, 12, 0, 4, TimeSpan.Zero),
+                GetLastEventReceivedAt(_client.InternalLowLevelClient));
+        }
+
+        [Test]
+        public void when_disconnect_mid_paced_recovery_expect_no_state_recovered()
+        {
+            _client.InternalLowLevelClient.EventDrainCountCap = 1;
+            _client.InternalLowLevelClient.EventDrainTimeBudgetMs = 10_000;
+
+            Connect();
+            WatchChannel("messaging:a");
+            RespondWith(QueryChannelsEndpoint, QueryChannelsJson("messaging:a", "a"));
+
+            ReconnectWithSync(ManySyncMessages("messaging:a", count: 5));
+            Assert.AreEqual(0, _recoveredEvents.Count);
+            AssertQueryChannelsCallCount(0);
+
+            DropConnection();
+            Update();
+            Update();
+
+            Assert.AreEqual(0, _recoveredEvents.Count);
+            AssertQueryChannelsCallCount(0);
+        }
+
         [Test]
         public void when_silent_recovery_syncs_custom_event_expect_channel_custom_event_received()
         {
@@ -624,6 +713,26 @@ private static string CustomEventJson(string cid, string type)
         private static string SyncEventsJson(params string[] events)
             => "{\"events\":[" + string.Join(",", events) + "]}";
 
+        private static string ManySyncMessages(string cid, int count)
+        {
+            var start = new DateTimeOffset(2026, 8, 24, 12, 0, 0, TimeSpan.Zero);
+            var events = new string[count];
+            for (var i = 0; i < count; i++)
+            {
+                events[i] = MessageNewJson(cid, $"msg-{i}", start.AddSeconds(i));
+            }
+
+            return SyncEventsJson(events);
+        }
+
+        private static DateTimeOffset? GetLastEventReceivedAt(StreamChatLowLevelClient client)
+        {
+            var field = typeof(StreamChatLowLevelClient).GetField("_lastEventReceivedAt",
+                BindingFlags.Instance | BindingFlags.NonPublic);
+            Assert.IsNotNull(field, "Expected _lastEventReceivedAt to exist.");
+            return (DateTimeOffset?)field.GetValue(client);
+        }
+
         private void ReconnectWithSync(string syncJson)
         {
             var now = new DateTimeOffset(2026, 8, 24, 12, 0, 0, TimeSpan.Zero);
diff --git a/Assets/Plugins/StreamChat/Tests/StateSync/StateSyncCatchUpTests.cs b/Assets/Plugins/StreamChat/Tests/StateSync/StateSyncCatchUpTests.cs
index 0642ccfb..a39e41ea 100644
--- a/Assets/Plugins/StreamChat/Tests/StateSync/StateSyncCatchUpTests.cs
+++ b/Assets/Plugins/StreamChat/Tests/StateSync/StateSyncCatchUpTests.cs
@@ -1,6 +1,7 @@
 #if STREAM_TESTS_ENABLED
 using System;
 using System.Reflection;
+using System.Threading.Tasks;
 using NSubstitute;
 using NUnit.Framework;
 using StreamChat.Core.Configs;
@@ -115,8 +116,7 @@ public void when_sync_replays_events_older_than_watermark_expect_watermark_not_r
             SetDisconnectionLastEventReceivedAt(now.AddHours(-1));
             StubSyncResponseWithMessageEvent(now.AddHours(-1));
 
-            _lowLevelClient.FetchAndProcessEventsSinceLastReceivedEvent(new[] { TestChannelCid }).GetAwaiter()
-                .GetResult();
+            FetchAndPumpUntilDone();
 
             Assert.AreEqual(now, GetLastEventReceivedAt());
         }
@@ -132,8 +132,7 @@ public void when_sync_replays_events_newer_than_watermark_expect_watermark_advan
             SetDisconnectionLastEventReceivedAt(now.AddHours(-2));
             StubSyncResponseWithMessageEvent(replayedEventCreatedAt);
 
-            _lowLevelClient.FetchAndProcessEventsSinceLastReceivedEvent(new[] { TestChannelCid }).GetAwaiter()
-                .GetResult();
+            FetchAndPumpUntilDone();
 
             Assert.AreEqual(replayedEventCreatedAt, GetLastEventReceivedAt());
         }
@@ -177,6 +176,20 @@ private void SetLastEventReceivedAt(DateTimeOffset value)
         private DateTimeOffset? GetLastEventReceivedAt()
             => (DateTimeOffset?)GetPrivateField("_lastEventReceivedAt").GetValue(_lowLevelClient);
 
+        private void FetchAndPumpUntilDone()
+        {
+            var task = _lowLevelClient.FetchAndProcessEventsSinceLastReceivedEvent(new[] { TestChannelCid });
+            var frames = 0;
+            while (!task.IsCompleted && frames < 1000)
+            {
+                _lowLevelClient.Update(0.1f);
+                frames++;
+            }
+
+            Assert.IsTrue(task.IsCompleted, "FetchAndProcessEventsSinceLastReceivedEvent did not finish after Update pumping.");
+            task.GetAwaiter().GetResult();
+        }
+
         private static FieldInfo GetPrivateField(string name)
         {
             var field = typeof(StreamChatLowLevelClient).GetField(name,