From c54721e637ca9119524c0f8f19bd5eb7f5c541ea Mon Sep 17 00:00:00 2001
From: Rene Floor <rene.floor@getstream.io>
Date: Thu, 6 Aug 2026 12:35:16 +0200
Subject: [PATCH 1/7] feat(llc): add ChannelClientState.isMarkedAsUnread
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Tracks whether the current user has an active manual mark-unread on
the channel that hasn't been read past yet, mirroring the iOS SDK's
ReadStateHandler.isMarkedAsUnread.

Set by markUnreadLocally and by a notification.mark_unread event for
the current user; cleared by markReadLocally and by a message.read
event for the current user.

Intended for UI-layer gating that shouldn't immediately undo a manual
mark-unread — used by stream_chat_flutter's tightened mark-read gating
(FLU-640).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
---
 packages/stream_chat/CHANGELOG.md             |  1 +
 .../stream_chat/lib/src/client/channel.dart   | 28 +++++++++++++++++--
 2 files changed, 27 insertions(+), 2 deletions(-)

diff --git a/packages/stream_chat/CHANGELOG.md b/packages/stream_chat/CHANGELOG.md
index 83018bf1b6..ba8648024f 100644
--- a/packages/stream_chat/CHANGELOG.md
+++ b/packages/stream_chat/CHANGELOG.md
@@ -5,6 +5,7 @@
 - Added `StreamChatClient.isLocalUnreadCountEnabled` (default `false`). When enabled, channels that have read events disabled (e.g. livestream channel types) track their unread count locally, on-device: incoming messages increment it, hard-deleted messages decrement it, and `Channel.markRead` / `markUnread` / `markUnreadByTimestamp` update it locally without a network request — including `Read.lastReadMessageId`, so the unread divider and jump-to-unread button anchor to the right message. Channels that support read receipts are unaffected and keep relying on server-driven unread counts.
 - Added `Event.watcherCount`, exposing the server-provided `watcher_count` field on events (e.g. `user.watching.start`, `user.watching.stop`, `message.new`).
 - Added `StreamChatNetworkError.type` (a `StreamChatNetworkErrorType` capturing the transport failure kind — connection error, timeout, cancellation, etc.).
+- Added `ChannelClientState.isMarkedAsUnread`, reporting whether the current user has an active manual mark-unread on the channel that hasn't been read past yet. Set by `markUnreadLocally` and by a `notification.mark_unread` event for the current user; cleared by `markReadLocally` and by a `message.read` event for the current user.
 
 ⚠️ Deprecated
 
diff --git a/packages/stream_chat/lib/src/client/channel.dart b/packages/stream_chat/lib/src/client/channel.dart
index 6bacb60a94..783cd79a02 100644
--- a/packages/stream_chat/lib/src/client/channel.dart
+++ b/packages/stream_chat/lib/src/client/channel.dart
@@ -3473,10 +3473,12 @@ class ChannelClientState {
             updateRead([updatedRead]);
 
             // If the read event is from the current user, reconcile the
-            // channel delivery status with the updated read state.
+            // channel delivery status with the updated read state, and clear
+            // any pending manual mark-unread — the user has read past it.
             final currentUser = _client.state.currentUser;
             if (event.isFromUser(userId: currentUser?.id)) {
               _client.channelDeliveryReporter.reconcileDelivery([_channel]);
+              _isMarkedAsUnread = false;
             }
           },
         ),
@@ -3499,7 +3501,14 @@ class ChannelClientState {
               lastDeliveredMessageId: currentRead?.lastDeliveredMessageId,
             );
 
-            return updateRead([updatedRead]);
+            updateRead([updatedRead]);
+
+            // Only a mark-unread for the current user's own read state
+            // should gate this device's auto mark-read.
+            final currentUser = _client.state.currentUser;
+            if (event.isFromUser(userId: currentUser?.id)) {
+              _isMarkedAsUnread = true;
+            }
           },
         ),
       )
@@ -3662,6 +3671,18 @@ class ChannelClientState {
     return updateRead([existingUserRead.copyWith(unreadMessages: count)]);
   }
 
+  /// Whether the current user explicitly marked a message in this channel as
+  /// unread during this session, without having read past that boundary
+  /// since.
+  ///
+  /// Set by [markUnreadLocally] and by a `notification.mark_unread` event for
+  /// the current user; cleared by [markReadLocally] and by a `message.read`
+  /// event for the current user. Intended for UI-layer gating that shouldn't
+  /// immediately undo a manual mark-unread — mirrors the iOS SDK's
+  /// `ReadStateHandler.isMarkedAsUnread`.
+  bool get isMarkedAsUnread => _isMarkedAsUnread;
+  bool _isMarkedAsUnread = false;
+
   /// Marks the channel as read locally, without making a network request.
   ///
   /// Used for channels that track unread counts locally (see
@@ -3700,6 +3721,8 @@ class ChannelClientState {
     // locally can still have delivery receipts enabled. Mirrors what the
     // `message.read` event listener does for server-driven channels.
     _client.channelDeliveryReporter.reconcileDelivery([_channel]);
+
+    _isMarkedAsUnread = false;
   }
 
   /// Marks the channel as unread locally, without making a network request.
@@ -3738,6 +3761,7 @@ class ChannelClientState {
     final unread = messages.where((it) => MessageRules.canCountAsUnread(it, _channel)).length;
 
     unreadCount = unread;
+    _isMarkedAsUnread = true;
   }
 
   /// Counts the number of unread messages mentioning the current user.

From 6035a2c32ab0fc36303d79383160232cf81fcae0 Mon Sep 17 00:00:00 2001
From: Rene Floor <rene.floor@getstream.io>
Date: Thu, 6 Aug 2026 12:35:37 +0200
Subject: [PATCH 2/7] feat(core): add StreamChannel.openAtFirstUnread

Gates the existing auto-scroll-to-first-unread positioning behind an
opt-out flag on StreamChannel/StreamChannel.value, defaulting to true
so existing integrations keep today's behavior unchanged. Set to
false to always open a channel at the latest message instead, and
let the message list surface pre-existing unread via its divider and
jump-to-unread pill rather than by scrolling there automatically.

Updates the sample app's channel route to demonstrate the flag.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
---
 .../stream_chat_flutter_core/CHANGELOG.md     |  1 +
 .../lib/src/stream_channel.dart               | 68 ++++++++++++-------
 sample_app/lib/routes/app_routes.dart         |  1 +
 3 files changed, 45 insertions(+), 25 deletions(-)

diff --git a/packages/stream_chat_flutter_core/CHANGELOG.md b/packages/stream_chat_flutter_core/CHANGELOG.md
index abc81cf51d..eb414e7d3f 100644
--- a/packages/stream_chat_flutter_core/CHANGELOG.md
+++ b/packages/stream_chat_flutter_core/CHANGELOG.md
@@ -4,6 +4,7 @@
 
 - Added `StreamChannelState.retry()` to re-run a failed channel initialization, for use as the retry action in `StreamChannel.errorBuilder`.
 - Added `DefaultStreamChannelBuilders`, an inherited widget that supplies default loading and error builders to descendant `StreamChannel`s (resolved via `loadingBuilderOf`/`errorBuilderOf`).
+- Added `StreamChannel.openAtFirstUnread`, defaulting to `true` (preserving existing behavior). Set to `false` to always open a channel at the latest message, instead of scrolling to the first pre-existing unread message.
 
 🐞 Fixed
 
diff --git a/packages/stream_chat_flutter_core/lib/src/stream_channel.dart b/packages/stream_chat_flutter_core/lib/src/stream_channel.dart
index 1f6c64f2e1..5e4232f356 100644
--- a/packages/stream_chat_flutter_core/lib/src/stream_channel.dart
+++ b/packages/stream_chat_flutter_core/lib/src/stream_channel.dart
@@ -37,6 +37,7 @@ class StreamChannel extends StatefulWidget {
     required this.channel,
     this.showLoading = true,
     this.initialMessageId,
+    this.openAtFirstUnread = true,
     this.errorBuilder = _resolveErrorBuilder,
     this.loadingBuilder = _resolveLoadingBuilder,
   }) : _shouldPosition = true;
@@ -60,6 +61,7 @@ class StreamChannel extends StatefulWidget {
     required this.channel,
   }) : showLoading = false,
        initialMessageId = null,
+       openAtFirstUnread = true,
        errorBuilder = _resolveErrorBuilder,
        loadingBuilder = _resolveLoadingBuilder,
        _shouldPosition = false;
@@ -76,6 +78,18 @@ class StreamChannel extends StatefulWidget {
   /// If passed the channel will load from this particular message.
   final String? initialMessageId;
 
+  /// Whether the channel should open positioned at the first unread message
+  /// when it has pre-existing unread messages.
+  ///
+  /// Defaults to `true`, preserving the SDK's existing behaviour. Set to
+  /// `false` to always open at the latest message instead — the message
+  /// list then surfaces pre-existing unread via its unread divider and
+  /// jump-to-unread pill rather than by scrolling there automatically.
+  ///
+  /// Has no effect on [StreamChannel.value], which never repositions the
+  /// loaded window.
+  final bool openAtFirstUnread;
+
   /// Widget builder used while the channel is initialising.
   ///
   /// Defaults to a builder that resolves the nearest
@@ -856,33 +870,37 @@ class StreamChannelState extends State<StreamChannel> {
       return loadChannelAtMessage(initialMessageId);
     }
 
-    // Otherwise, we should load the channel at the first unread
-    // message if available.
-    if (channel.state case final state? when state.unreadCount > 0) {
-      final currentUserRead = state.currentUserRead;
-
-      // Skip if we don't have read state for the current user.
-      if (currentUserRead == null) return;
-
-      // Load the channel at the last read message if available.
-      if (currentUserRead.lastReadMessageId case final lastReadMessageId?) {
-        try {
-          return await loadChannelAtMessage(lastReadMessageId);
-        } catch (e) {
-          // If the loadChannelAtMessage for any reason fails, we fallback to
-          // loading the channel at the last read date.
-          //
-          // One example of this is when the channel becomes too large and
-          // exceeds a certain threshold (I believe it's a 1000 members) it
-          // can't update the readstate anymore for each individual member.
+    // Otherwise, we should load the channel at the first unread message if
+    // available — unless the caller opted out via
+    // [StreamChannel.openAtFirstUnread], in which case we fall through to
+    // load-latest below.
+    if (widget.openAtFirstUnread) {
+      if (channel.state case final state? when state.unreadCount > 0) {
+        final currentUserRead = state.currentUserRead;
+
+        // Skip if we don't have read state for the current user.
+        if (currentUserRead == null) return;
+
+        // Load the channel at the last read message if available.
+        if (currentUserRead.lastReadMessageId case final lastReadMessageId?) {
+          try {
+            return await loadChannelAtMessage(lastReadMessageId);
+          } catch (e) {
+            // If the loadChannelAtMessage for any reason fails, we fallback to
+            // loading the channel at the last read date.
+            //
+            // One example of this is when the channel becomes too large and
+            // exceeds a certain threshold (I believe it's a 1000 members) it
+            // can't update the readstate anymore for each individual member.
+          }
         }
-      }
 
-      // Skip the "never read" sentinel: the server ignores it as
-      // `created_at_around` and returns the tail, which would mis-infer
-      // `_topPaginationEnded = true`. Fall through to load-latest below.
-      if (currentUserRead.lastRead.isAfter(_minValidLastRead)) {
-        return loadChannelAtTimestamp(currentUserRead.lastRead);
+        // Skip the "never read" sentinel: the server ignores it as
+        // `created_at_around` and returns the tail, which would mis-infer
+        // `_topPaginationEnded = true`. Fall through to load-latest below.
+        if (currentUserRead.lastRead.isAfter(_minValidLastRead)) {
+          return loadChannelAtTimestamp(currentUserRead.lastRead);
+        }
       }
     }
 
diff --git a/sample_app/lib/routes/app_routes.dart b/sample_app/lib/routes/app_routes.dart
index 3e11a35f2f..ca603bd580 100644
--- a/sample_app/lib/routes/app_routes.dart
+++ b/sample_app/lib/routes/app_routes.dart
@@ -42,6 +42,7 @@ final appRoutes = [
           return StreamChannel(
             channel: channel,
             initialMessageId: messageId,
+            openAtFirstUnread: false,
             child: Builder(
               builder: (context) {
                 return (parentMessage != null)

From 2622109e3bfe046ed55d07aa2596604eb2e24529 Mon Sep 17 00:00:00 2001
From: Rene Floor <rene.floor@getstream.io>
Date: Thu, 6 Aug 2026 12:45:28 +0200
Subject: [PATCH 3/7] feat(ui): rework unread indicators and tighten mark-read
 gating (FLU-648/649/650/640)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Unread messages divider: anchored to the pre-existing read/unread
boundary captured when the channel opens. The anchor is frozen for
the whole session — it never moves or disappears, regardless of
scrolling or reads — but its displayed count keeps counting up as
further messages arrive during the session (mirroring WhatsApp)
instead of staying fixed at the open-time total.

Jump-to-unread pill (UnreadIndicatorButton): shows the frozen
open-time count, gated on that boundary sitting above the viewport.
Visible as soon as the count is known from the channel's Read state,
even before the boundary message itself has loaded — tapping it
before then falls back to loadChannelAtMessage via the boundary's
lastReadMessageId. Dismisses permanently for the session on tap, the
dismiss button, or scrolling past it; the button itself is now purely
presentational, taking a required unreadCount instead of subscribing
to read state internally.

Scroll-to-bottom badge: counts only messages that arrive while
scrolled away from the bottom (never seeded from the channel's unread
count, unlike the divider above), and always resets to 0 once the
user reaches the bottom.

Mark-read gating (FLU-640): tightened to mirror iOS's
shouldMarkChannelRead — besides isUpToDate and unreadCount > 0, now
also requires the bottom to have been seen (now, or earlier then
scrolled away), the pre-existing boundary (if any) to have been seen
or scrolled past, and no active manual mark-unread
(Channel.isMarkedAsUnread). That last check can't gate on the flag
directly and permanently: it only clears via a successful mark-read,
which is the very thing it would be gating, so it would deadlock the
channel unread forever the moment it's set. Instead it latches once
the viewport genuinely diverges from a snapshot taken when the
mark-unread was first observed — captured eagerly on a live
transition, or on the first laid-out frame as a fallback for a
channel that simply mounts already marked unread.

Adds StreamMessageListViewConfiguration.shouldMarkRead to override
this gating entirely, and Translations.unreadMessagesSeparatorLabel
(added rather than changing the existing unreadMessagesSeparatorText,
to avoid breaking existing overrides) so the default separator can
show a count.

Also defaults MockChannelState.isMarkedAsUnread to false, since
_handleItemPositionsChanged now reads it on every scroll tick and
existing test files that construct the mock without stubbing it
would otherwise crash.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
---
 packages/stream_chat_flutter/CHANGELOG.md     |  12 +
 .../lib/src/localization/translations.dart    |  10 +
 .../message_list_view/mark_read_details.dart  |  42 ++
 .../message_list_view/message_list_view.dart  | 489 ++++++++++++++----
 .../lib/src/message_list_view/mlv_utils.dart  |  17 +-
 ...tream_message_list_view_configuration.dart |  20 +-
 .../unread_indicator_button.dart              |  53 +-
 .../unread_messages_separator.dart            |   6 +-
 .../lib/stream_chat_flutter.dart              |   2 +
 .../default_translations_test.dart            |   2 +
 .../src/message_list_view/mark_read_test.dart | 366 ++++++++++++-
 .../unread_divider_test.dart                  | 429 +++++++++++++++
 .../stream_chat_flutter/test/src/mocks.dart   |   1 +
 13 files changed, 1298 insertions(+), 151 deletions(-)
 create mode 100644 packages/stream_chat_flutter/lib/src/message_list_view/mark_read_details.dart
 create mode 100644 packages/stream_chat_flutter/test/src/message_list_view/unread_divider_test.dart

diff --git a/packages/stream_chat_flutter/CHANGELOG.md b/packages/stream_chat_flutter/CHANGELOG.md
index 2220c44b44..f5814518bd 100644
--- a/packages/stream_chat_flutter/CHANGELOG.md
+++ b/packages/stream_chat_flutter/CHANGELOG.md
@@ -8,6 +8,18 @@
 - Added a `size` (`StreamLoadingSpinnerSize`) parameter to `StreamScrollViewLoadingWidget`.
 - Added `onReactionTap` to `StreamMessageItem` and `StreamMessageListView`, reporting the tapped message's `BuildContext` and a `ReactionTapDetails` with the tapped `message` and `reaction` (the reaction is `null` for a clustered or overflow chip that maps to no single reaction).
 - Added an `unreadIndicator` parameter to `StreamBackButton` that overlays a widget (typically a `StreamUnreadIndicator`) on the button's top-end corner. Pass `StreamUnreadIndicator(excludeCid: cid)` to show the total unread count of other channels, or `StreamUnreadIndicator.channels(cid: cid)` for a single channel's count.
+- Added `StreamMessageListViewConfiguration.shouldMarkRead` to fully override the automatic mark-read gating described below.
+- Added `Channel.isMarkedAsUnread` (via `ChannelClientState`), reporting whether the current user has an active manual mark-unread that hasn't been read past yet.
+- Added `StreamChannel.openAtFirstUnread` (`stream_chat_flutter_core`), defaulting to `true`. Set to `false` to always open a channel at the latest message instead of scrolling to the first pre-existing unread message.
+- Added `Translations.unreadMessagesSeparatorLabel`, used by the default `UnreadMessagesSeparator` to show a count, e.g. "5 unread messages".
+
+🔄 Changed
+
+- Changed the "↑ N unread" jump-to-unread pill to a count fixed when the channel opens, staying on screen for the whole session rather than reacting to the live, shrinking unread count. The pill now shows as soon as that count is known — even before the boundary message itself has loaded — and dismisses permanently for the session once tapped, dismissed, or scrolled past; it no longer reappears when a new message arrives.
+- Changed the scroll-to-bottom badge to count only messages that arrive out of view during the current session, rather than being seeded from the channel's unread count. It always resets to 0 once the user reaches the bottom.
+- Changed the "unread messages" divider to show a count, starting at the channel's open-time unread total and counting up as further messages arrive during the session — mirroring WhatsApp — instead of a fixed, count-less label.
+- Changed `UnreadIndicatorButton` to take a `required int unreadCount` and render unconditionally, dropping its internal read-state subscription — `StreamMessageListView` now owns its visibility.
+- Tightened `StreamMessageListView`'s automatic mark-read gating to also require that the pre-existing unread boundary (if any) has been seen or scrolled past, and that there's no pending manual mark-unread — mirroring the iOS SDK. Previously, reaching the bottom with unread messages present was sufficient.
 
 ⚠️ Deprecated
 
diff --git a/packages/stream_chat_flutter/lib/src/localization/translations.dart b/packages/stream_chat_flutter/lib/src/localization/translations.dart
index d495e6564f..0a96e78e6b 100644
--- a/packages/stream_chat_flutter/lib/src/localization/translations.dart
+++ b/packages/stream_chat_flutter/lib/src/localization/translations.dart
@@ -102,6 +102,10 @@ abstract class Translations {
   /// in the [StreamMessageListView]
   String unreadMessagesSeparatorText();
 
+  /// The label for the unread messages separator in the
+  /// [StreamMessageListView], e.g. "5 unread messages".
+  String unreadMessagesSeparatorLabel({required int count});
+
   /// The label for "connected" in [StreamConnectionStatusBuilder]
   String get connectedLabel;
 
@@ -1290,6 +1294,12 @@ Attachment limit exceeded: it's not possible to add more than $limit attachments
   @override
   String unreadMessagesSeparatorText() => 'New messages';
 
+  @override
+  String unreadMessagesSeparatorLabel({required int count}) {
+    if (count == 1) return '1 unread message';
+    return '$count unread messages';
+  }
+
   @override
   String get enableFileAccessMessage =>
       'Please enable access to files'
diff --git a/packages/stream_chat_flutter/lib/src/message_list_view/mark_read_details.dart b/packages/stream_chat_flutter/lib/src/message_list_view/mark_read_details.dart
new file mode 100644
index 0000000000..d2b4fda27d
--- /dev/null
+++ b/packages/stream_chat_flutter/lib/src/message_list_view/mark_read_details.dart
@@ -0,0 +1,42 @@
+/// The information available when deciding whether to automatically mark a
+/// [StreamMessageListView]'s channel as read.
+///
+/// Passed to a caller-supplied predicate on
+/// [StreamMessageListViewConfiguration.shouldMarkRead]. Not intended to be
+/// constructed directly.
+class StreamMarkReadDetails {
+  /// Creates a set of details describing the current mark-read gate state.
+  const StreamMarkReadDetails({
+    required this.hasSeenLastMessage,
+    required this.hasSeenFirstUnreadMessage,
+    required this.isMarkedAsUnread,
+    required this.unreadCount,
+  });
+
+  /// Whether the bottom of the list has been fully visible at some point
+  /// since the last successful mark-read — either it's visible right now, or
+  /// it was visible earlier and the user has since scrolled away.
+  final bool hasSeenLastMessage;
+
+  /// Whether the user has seen (rendered on screen) or scrolled past the
+  /// pre-existing unread boundary captured when the channel was opened.
+  ///
+  /// Always `true` when there was nothing to see in the first place — the
+  /// channel opened fully read, or it uses local unread counts with read
+  /// events disabled.
+  final bool hasSeenFirstUnreadMessage;
+
+  /// Whether the current user has an active manual mark-unread on this
+  /// channel that hasn't been read past yet.
+  final bool isMarkedAsUnread;
+
+  /// The channel's current unread count.
+  final int unreadCount;
+}
+
+/// Signature for overriding [StreamMessageListView]'s automatic mark-read
+/// gating.
+///
+/// Return `true` to mark the channel as read, `false` to skip it for now —
+/// the list retries on the next relevant scroll or message event.
+typedef StreamShouldMarkReadPredicate = bool Function(StreamMarkReadDetails details);
diff --git a/packages/stream_chat_flutter/lib/src/message_list_view/message_list_view.dart b/packages/stream_chat_flutter/lib/src/message_list_view/message_list_view.dart
index c8ecf79d75..e84be35481 100644
--- a/packages/stream_chat_flutter/lib/src/message_list_view/message_list_view.dart
+++ b/packages/stream_chat_flutter/lib/src/message_list_view/message_list_view.dart
@@ -12,7 +12,6 @@ import 'package:stream_chat_flutter/src/message_list_view/mlv_utils.dart';
 import 'package:stream_chat_flutter/src/message_list_view/stream_message_list_empty_state.dart';
 import 'package:stream_chat_flutter/src/message_list_view/stream_message_list_skeleton_loading.dart';
 import 'package:stream_chat_flutter/src/message_list_view/thread_separator.dart';
-import 'package:stream_chat_flutter/src/message_list_view/unread_messages_separator.dart';
 import 'package:stream_chat_flutter/src/message_widget/stream_ephemeral_message.dart';
 import 'package:stream_chat_flutter/src/misc/empty_widget.dart';
 import 'package:stream_chat_flutter/src/utils/network_error_text.dart';
@@ -295,17 +294,144 @@ class _StreamMessageListViewState extends State<StreamMessageListView> {
   late final ItemPositionsListener _itemPositionListener;
   StreamChannelState? streamChannel;
 
-  // Drives the unread-messages separator. Held in a [ValueNotifier] so read
-  // events can update it without rebuilding the entire list view.
-  final _unreadState = ValueNotifier<({int count, String? firstUnreadId})>((count: 0, firstUnreadId: null));
+  // --- Divider A: pre-existing unread, frozen at channel open ---
+  //
+  // [_unreadBaseline] is the current user's [Read] captured once when the
+  // channel is attached (or on the first `currentUserReadStream` emission if
+  // read state wasn't available yet). It never changes afterwards, so
+  // resolving the anchor against it — rather than against the live,
+  // ever-shrinking `unreadCount` — is what keeps the divider and pill on
+  // screen across an auto mark-read.
+  Read? _unreadBaseline;
+  bool _unreadBaselineCaptured = false;
+
+  // Resolved anchor for divider A. The anchor (and `count`, the frozen
+  // baseline used by the pill) is frozen once non-null: recomputation is
+  // skipped as soon as `anchorId` is set. May take a few rebuilds to resolve
+  // if top pagination hasn't finished loading the boundary yet.
+  final _unreadDivider = ValueNotifier<({int count, String? anchorId})>((count: 0, anchorId: null));
+
+  // Grows by one for every message that arrives out of view while divider
+  // A is on screen, so the divider's displayed count keeps counting up
+  // during the session (mirroring WhatsApp) instead of staying frozen at
+  // the open-time count. Added on top of `_unreadDivider.value.count` for
+  // display only — the pill keeps using the frozen count.
+  final ValueNotifier<int> _unreadDividerGrowth = ValueNotifier(0);
+
+  // Sticky: becomes true once the user has seen (rendered) or scrolled past
+  // divider A's anchor. Drives the pill's permanent dismissal and (see
+  // [_maybeMarkMessagesAsRead]) gates auto mark-read.
+  final ValueNotifier<bool> _hasSeenFirstUnread = ValueNotifier(false);
+
+  // Scroll-to-bottom badge count. Counts messages that arrive while the user
+  // is scrolled away from the bottom; resets to 0 once they reach the bottom.
+  final ValueNotifier<int> _scrollToBottomBadge = ValueNotifier(0);
+
+  // Sticky "bottom was reached" flag for the FLU-640 mark-read gate. Cleared
+  // after each successful mark-read so returning to the bottom is required
+  // again before the next one.
+  bool _hasSeenLastMessage = false;
+
+  // While non-null, the viewport captured at the moment an active manual
+  // mark-unread (`channel.state.isMarkedAsUnread`) was first observed.
+  // `_maybeMarkMessagesAsRead` blocks until [_markUnreadViewportDiverged]
+  // is true — evidence the user did something (scrolled, reopened the
+  // channel, etc.) since marking the message unread, rather than the
+  // anchor merely being immediately "visible" again because it's usually
+  // the very message just marked and nothing has moved.
+  //
+  // This can't gate on `isMarkedAsUnread` directly and permanently: that
+  // flag only clears via a successful mark-read, which is the very thing
+  // it would be gating, so treating it as a persistent block would
+  // deadlock the channel unread forever the moment it's set — the exact
+  // bug this snapshot exists to avoid.
+  //
+  // Set eagerly in [_handleCurrentUserReadChanged] right when a live
+  // transition is observed (captures the precise pre-scroll viewport), and
+  // in [_handleItemPositionsChanged] on the first genuinely laid-out frame
+  // as a fallback for when the channel simply mounts with
+  // `isMarkedAsUnread` already true and no transition ever fires — that
+  // has to happen there and not lazily inside [_maybeMarkMessagesAsRead],
+  // since the first time that gate is evaluated might already be the
+  // user's first genuine arrival at the bottom, which would otherwise be
+  // burned on capturing the baseline instead of acting on it. Cleared once
+  // a mark-read actually goes through, or once `isMarkedAsUnread` itself
+  // clears (so a future mark-unread starts its own fresh snapshot).
+  Iterable<ItemPosition>? _markUnreadViewportSnapshot;
+
+  // Sticky once true: sighted the first time [_handleItemPositionsChanged]
+  // (or, as a fallback, [_maybeMarkMessagesAsRead] itself) sees item
+  // positions that genuinely differ from [_markUnreadViewportSnapshot].
+  // Deliberately tracked as "did this ever happen" rather than
+  // re-comparing the *current* positions against the snapshot on each
+  // check — a user who scrolls away and back settles at the exact same
+  // rest position, which would otherwise look unchanged and re-block a
+  // mark-read that should already have been earned by that round trip.
+  bool _markUnreadViewportDiverged = false;
+
+  // Captures [_unreadBaseline] the first time the current user's read state
+  // becomes available, then attempts to resolve divider A's anchor against
+  // it. No-ops in a thread, where divider A doesn't apply.
+  void _captureUnreadBaselineIfNeeded() {
+    if (_unreadBaselineCaptured || _isThreadConversation) return;
+
+    final currentUserRead = streamChannel?.channel.state?.currentUserRead;
+    if (currentUserRead == null) return;
+
+    _unreadBaselineCaptured = true;
+    _unreadBaseline = currentUserRead.unreadMessages > 0 ? currentUserRead : null;
+    // Publish the frozen count right away, even though the anchor itself
+    // can't resolve until top pagination has loaded that far back — the
+    // pill only needs the count, not the anchor, so it shouldn't wait on
+    // pagination to appear (see `_onUnreadPillJumpTap` for how a tap
+    // before the anchor resolves still jumps there).
+    if (_unreadBaseline case final baseline?) {
+      _unreadDivider.value = (count: baseline.unreadMessages, anchorId: _unreadDivider.value.anchorId);
+    }
+    _resolveUnreadDivider();
+  }
 
-  // Snapshot of the current user's unread state, sourced from the channel. Used
-  // both to seed [_unreadState] on channel attach and to refresh it from the
-  // [Channel.currentUserReadStream] listener.
-  ({int count, String? firstUnreadId}) _readUnreadSnapshot() => (
-    count: streamChannel?.channel.state?.unreadCount ?? 0,
-    firstUnreadId: streamChannel?.getFirstUnreadMessage()?.id,
-  );
+  // Resolves divider A's anchor against the frozen baseline. A no-op once
+  // resolved, and while top pagination hasn't loaded the boundary yet.
+  void _resolveUnreadDivider() {
+    if (_isThreadConversation || _unreadDivider.value.anchorId != null) return;
+
+    final baseline = _unreadBaseline;
+    if (baseline == null) return;
+
+    final anchor = streamChannel?.getFirstUnreadMessage(baseline);
+    if (anchor == null) return;
+
+    _unreadDivider.value = (count: baseline.unreadMessages, anchorId: anchor.id);
+  }
+
+  // Reacts to a `currentUserReadStream` emission. An explicit mark-unread
+  // moves the read boundary backward — treat it as a new session start for
+  // divider A/the pill, mirroring iOS's `forceUpdate` path.
+  void _handleCurrentUserReadChanged() {
+    if (_isThreadConversation) return;
+
+    final channel = streamChannel?.channel;
+    if (channel == null) return;
+
+    if (channel.state?.isMarkedAsUnread ?? false) {
+      _unreadBaselineCaptured = false;
+      _unreadBaseline = null;
+      _unreadDivider.value = (count: 0, anchorId: null);
+      _unreadDividerGrowth.value = 0;
+      _hasSeenFirstUnread.value = false;
+      // Only capture once per mark-unread session — a later, unrelated
+      // read-stream emission while still marked unread shouldn't keep
+      // chasing the latest position and never let a genuine scroll differ
+      // from it.
+      _markUnreadViewportSnapshot ??= _itemPositionListener.itemPositions.value.toList();
+    } else {
+      _markUnreadViewportSnapshot = null;
+      _markUnreadViewportDiverged = false;
+    }
+
+    _captureUnreadBaselineIfNeeded();
+  }
 
   bool get _upToDate => streamChannel!.channel.state!.isUpToDate;
 
@@ -361,7 +487,16 @@ class _StreamMessageListViewState extends State<StreamMessageListView> {
       debouncedMarkRead.cancel();
       debouncedMarkThreadRead.cancel();
 
-      _unreadState.value = _readUnreadSnapshot();
+      _unreadBaselineCaptured = false;
+      _unreadBaseline = null;
+      _unreadDivider.value = (count: 0, anchorId: null);
+      _unreadDividerGrowth.value = 0;
+      _scrollToBottomBadge.value = 0;
+      _hasSeenFirstUnread.value = false;
+      _hasSeenLastMessage = false;
+      _markUnreadViewportSnapshot = null;
+      _markUnreadViewportDiverged = false;
+      _captureUnreadBaselineIfNeeded();
 
       final highlightInitialMessage = widget.config.highlightInitialMessage;
       final highlightMessageId = switch ((highlightInitialMessage, _isThreadConversation)) {
@@ -392,6 +527,21 @@ class _StreamMessageListViewState extends State<StreamMessageListView> {
         final currentUser = streamChannel?.channel.client.state.currentUser;
         final isAtBottom = !_showScrollToBottom.value;
 
+        // The scroll-to-bottom badge and divider A's growing count only
+        // apply to the channel's own message stream (not thread replies),
+        // and never count the current user's own messages.
+        final isOwnMessage = message.user?.id == currentUser?.id;
+        if (!_isThreadConversation && !isOwnMessage) {
+          // The divider counts every qualifying arrival — including ones
+          // seen live at the bottom — so it keeps counting up like
+          // WhatsApp's. The badge is narrower: it only exists to flag
+          // what was missed while scrolled away, so it skips arrivals
+          // that were already in view and resets once the bottom is
+          // reached (see `_handleItemPositionsChanged`).
+          _unreadDividerGrowth.value += 1;
+          if (!isAtBottom) _scrollToBottomBadge.value += 1;
+        }
+
         final details = StreamAutoScrollDetails(
           message: message,
           currentUser: currentUser,
@@ -417,14 +567,14 @@ class _StreamMessageListViewState extends State<StreamMessageListView> {
 
       _userReadListener?.cancel();
       _userReadListener = state?.currentUserReadStream.listen((_) {
-        _unreadState.value = _readUnreadSnapshot();
+        _handleCurrentUserReadChanged();
       });
     }
   }
 
   @override
   void dispose() {
-    // Tear down anything that could write to [_unreadState] or
+    // Tear down anything that could write to the unread/badge notifiers or
     // [_showScrollToBottom] before disposing them.
     _messageNewListener?.cancel();
     _messageNewListener = null;
@@ -433,7 +583,10 @@ class _StreamMessageListViewState extends State<StreamMessageListView> {
     _itemPositionListener.itemPositions.removeListener(_handleItemPositionsChanged);
     debouncedMarkRead.cancel();
     debouncedMarkThreadRead.cancel();
-    _unreadState.dispose();
+    _unreadDivider.dispose();
+    _unreadDividerGrowth.dispose();
+    _hasSeenFirstUnread.dispose();
+    _scrollToBottomBadge.dispose();
     _highlightState.dispose();
     super.dispose();
   }
@@ -591,6 +744,17 @@ class _StreamMessageListViewState extends State<StreamMessageListView> {
   Widget _buildListView(List<Message> data) {
     messages = data;
 
+    // Top pagination may not have finished loading the unread boundary when
+    // the baseline was first captured; retry once this frame's layout
+    // settles. Deferred (not synchronous) since mutating a [ValueNotifier]
+    // read by a [ValueListenableBuilder] further down this same build would
+    // notify a listener that hasn't rebuilt yet this frame.
+    if (_unreadBaseline != null && _unreadDivider.value.anchorId == null) {
+      WidgetsBinding.instance.addPostFrameCallback((_) {
+        if (mounted) _resolveUnreadDivider();
+      });
+    }
+
     final itemCount =
         messages.length + // total messages
         2 + // top + bottom loading indicator
@@ -816,9 +980,30 @@ class _StreamMessageListViewState extends State<StreamMessageListView> {
         if (widget.config.showUnreadIndicator && !_isThreadConversation)
           Positioned(
             top: context.streamSpacing.sm,
-            child: UnreadIndicatorButton(
-              onJumpTap: scrollToUnreadDefaultTapAction,
-              onDismissTap: _markMessagesAsRead,
+            child: ValueListenableBuilder(
+              valueListenable: _unreadDivider,
+              builder: (context, unread, _) {
+                // Gated on the frozen count, not the anchor: the count is
+                // known immediately from the baseline `Read`, while the
+                // anchor can take a while longer to resolve if top
+                // pagination hasn't loaded that far back yet. Waiting for
+                // the anchor would mean the pill — the whole point of
+                // which is to point at unread content the user hasn't
+                // scrolled to — only appeared once they'd already
+                // scrolled most of the way there themselves.
+                if (unread.count <= 0) return const Empty();
+                return ValueListenableBuilder<bool>(
+                  valueListenable: _hasSeenFirstUnread,
+                  builder: (context, seen, __) {
+                    if (seen) return const Empty();
+                    return UnreadIndicatorButton(
+                      unreadCount: unread.count,
+                      onJumpTap: _onUnreadPillJumpTap,
+                      onDismissTap: _onUnreadPillDismissTap,
+                    );
+                  },
+                );
+              },
             ),
           ),
       ],
@@ -861,33 +1046,32 @@ class _StreamMessageListViewState extends State<StreamMessageListView> {
   }
 
   Widget _buildUnreadMessagesSeparator(int unreadCount) {
-    if (widget.builders.unreadMessagesSeparator != null) {
-      return widget.builders.unreadMessagesSeparator!(context, unreadCount);
+    if (widget.builders.unreadMessagesSeparator case final builder?) {
+      return builder(context, unreadCount);
     }
     return UnreadMessagesSeparator(unreadCount: unreadCount);
   }
 
   // Wraps an already-built [separator] with the unread-messages line if
-  // [message] happens to be the first unread one. Defined as a method
-  // (rather than a closure inside [separatorBuilder]) so a fresh inner
-  // closure isn't allocated for every visible separator on every rebuild.
+  // [message] happens to be divider A's anchor. Defined as a method (rather
+  // than a closure inside [separatorBuilder]) so a fresh inner closure isn't
+  // allocated for every visible separator on every rebuild.
   Widget _maybeBuildWithUnreadMessagesSeparator({
     required Message message,
     required Widget separator,
   }) {
     if (_isThreadConversation) return separator;
     return ValueListenableBuilder(
-      valueListenable: _unreadState,
-      builder: (context, state, _) {
-        if (state.count == 0) return separator;
-        if (state.firstUnreadId != message.id) return separator;
-        return Column(
-          mainAxisSize: MainAxisSize.min,
-          crossAxisAlignment: CrossAxisAlignment.stretch,
-          children: [
-            separator,
-            _buildUnreadMessagesSeparator(state.count),
-          ],
+      valueListenable: _unreadDivider,
+      builder: (context, unread, _) {
+        if (unread.anchorId != message.id) return separator;
+        return ValueListenableBuilder<int>(
+          valueListenable: _unreadDividerGrowth,
+          builder: (context, growth, __) => Column(
+            mainAxisSize: MainAxisSize.min,
+            crossAxisAlignment: CrossAxisAlignment.stretch,
+            children: [separator, _buildUnreadMessagesSeparator(unread.count + growth)],
+          ),
         );
       },
     );
@@ -913,20 +1097,25 @@ class _StreamMessageListViewState extends State<StreamMessageListView> {
     }
   }
 
-  Future<void> scrollToUnreadDefaultTapAction(String? lastReadMessageId) async {
-    final firstUnreadId = _unreadState.value.firstUnreadId;
-    if (firstUnreadId == null) return;
-
-    // Scroll to the first unread message in the list.
-    final firstUnreadMessageIndex = messages.lastIndexWhere((it) => it.id == firstUnreadId);
-    if (firstUnreadMessageIndex == -1) return;
+  Future<void> _onUnreadPillJumpTap() async {
+    // The anchor may not have resolved yet if top pagination hasn't loaded
+    // that far back — the pill is visible already (see its gating above),
+    // so fall back to the frozen baseline's own last-read boundary, known
+    // immediately from the server `Read`, rather than doing nothing.
+    final anchorId = _unreadDivider.value.anchorId ?? _unreadBaseline?.lastReadMessageId;
+    if (anchorId == null) return;
+
+    _hasSeenFirstUnread.value = true;
+    // Delegates to [_scrollToMessage], which falls back to
+    // [StreamChannelState.loadChannelAtMessage] when the anchor isn't in the
+    // currently loaded window — after which the real anchor resolves
+    // naturally via the retry in [_buildListView], rendering divider A too.
+    await _scrollToMessage(messageId: anchorId, highlight: false);
+  }
 
-    if (_scrollController case final controller? when controller.isAttached) {
-      return controller.scrollTo(
-        index: max(firstUnreadMessageIndex + 2, 0),
-        alignment: 0.5, // center the message in the viewport
-      );
-    }
+  Future<void> _onUnreadPillDismissTap() async {
+    _hasSeenFirstUnread.value = true;
+    await _markMessagesAsRead();
   }
 
   late final debouncedMarkRead = debounce(
@@ -1062,15 +1251,14 @@ class _StreamMessageListViewState extends State<StreamMessageListView> {
   }
 
   Widget _buildScrollToBottom() {
-    return ValueListenableBuilder(
-      valueListenable: _unreadState,
-      builder: (_, state, __) {
-        final unreadCount = state.count;
+    return ValueListenableBuilder<int>(
+      valueListenable: _scrollToBottomBadge,
+      builder: (_, badgeCount, __) {
         if (widget.builders.scrollToBottomButton case final builder?) {
-          return builder(unreadCount, scrollToBottomDefaultTapAction);
+          return builder(badgeCount, scrollToBottomDefaultTapAction);
         }
 
-        final showUnreadCount = unreadCount > 0;
+        final showUnreadCount = badgeCount > 0;
 
         Widget button = StreamButton.icon(
           style: .secondary,
@@ -1081,12 +1269,12 @@ class _StreamMessageListViewState extends State<StreamMessageListView> {
             true => Icon(context.streamIcons.arrowDown),
             false => Icon(context.streamIcons.arrowUp),
           },
-          onPressed: () => scrollToBottomDefaultTapAction(unreadCount),
+          onPressed: () => scrollToBottomDefaultTapAction(badgeCount),
         );
 
         if (showUnreadCount && widget.config.showUnreadCountOnScrollToBottom) {
           button = StreamBadgeNotification(
-            label: '${unreadCount > 99 ? '99+' : unreadCount}',
+            label: '${badgeCount > 99 ? '99+' : badgeCount}',
             child: button,
           );
         }
@@ -1197,6 +1385,23 @@ class _StreamMessageListViewState extends State<StreamMessageListView> {
     final itemPositions = _itemPositionListener.itemPositions.value;
     if (itemPositions.isEmpty) return;
 
+    // Snapshot the viewport (or check it against an existing snapshot for
+    // divergence) the first time it's genuinely laid out while marked as
+    // unread, in case the channel simply mounted in that state rather than
+    // [_handleCurrentUserReadChanged] observing a live transition to hook
+    // the snapshot on. Doing this here — on every non-empty layout, before
+    // checking anything else below — rather than lazily inside
+    // [_maybeMarkMessagesAsRead], matters: that gate is only ever evaluated
+    // when a mark-read could fire, which for a channel the user opens and
+    // immediately scrolls all the way through might be the very first time
+    // they reach the bottom. Capturing the baseline there would burn that
+    // first genuine read on the snapshot itself instead of acting on it.
+    if (streamChannel?.channel.state?.isMarkedAsUnread ?? false) {
+      _checkMarkUnreadViewportDivergence(itemPositions);
+    }
+
+    final justSeenFirstUnread = _maybeUpdateHasSeenFirstUnread(itemPositions);
+
     // Index of the last item in the list view is 2 as 1 is the progress
     // indicator and 0 is the footer.
     const lastItemIndex = 2;
@@ -1212,61 +1417,163 @@ class _StreamMessageListViewState extends State<StreamMessageListView> {
     }
 
     if (mounted) _showScrollToBottom.value = !isLastItemFullyVisible;
-    if (isLastItemFullyVisible) return _handleLastItemFullyVisible();
-  }
-
-  Message? _lastFullyVisibleMessage;
-  void _handleLastItemFullyVisible() {
-    // We are using the first message as the last fully visible message
-    // because the messages are reversed in the list view.
-    final newLastFullyVisibleMessage = messages.firstOrNull;
+    if (isLastItemFullyVisible) {
+      _hasSeenLastMessage = true;
+      _scrollToBottomBadge.value = 0;
+    }
 
-    final lastFullyVisibleMessageChanged = switch (_lastFullyVisibleMessage) {
-      final message? => message.id != newLastFullyVisibleMessage?.id,
-      null => true, // Allows setting the initial value.
-    };
+    // Attempt a mark-read whenever either half of the FLU-640 gate could
+    // have just become satisfied; `_maybeMarkMessagesAsRead` does the actual
+    // deciding, and the leading-edge debounce inside it makes repeated
+    // attempts cheap.
+    if ((isLastItemFullyVisible || justSeenFirstUnread) && widget.config.markReadWhenAtTheBottom) {
+      _maybeMarkMessagesAsRead().ignore();
+    }
+  }
 
-    // If the last fully visible message has been changed, we need to update the
-    // value and maybe mark messages as read if needed.
-    if (lastFullyVisibleMessageChanged) {
-      _lastFullyVisibleMessage = newLastFullyVisibleMessage;
+  // Captures [_markUnreadViewportSnapshot] the first time this is called,
+  // and otherwise checks [itemPositions] against it, latching
+  // [_markUnreadViewportDiverged] the first time they genuinely differ.
+  // Deliberately latching rather than re-comparing *current* positions
+  // against the snapshot on every check: a user who scrolls away and back
+  // settles at the exact same rest position, which would otherwise look
+  // unchanged and re-block a mark-read the round trip should already have
+  // earned. Safe to call on every position-changed tick — a no-op once
+  // already diverged.
+  void _checkMarkUnreadViewportDivergence(Iterable<ItemPosition> itemPositions) {
+    if (_markUnreadViewportSnapshot == null) {
+      _markUnreadViewportSnapshot = itemPositions.toList();
+      return;
+    }
+    if (_markUnreadViewportDiverged) return;
 
-      // Mark messages as read if needed.
-      if (widget.config.markReadWhenAtTheBottom) {
-        _maybeMarkMessagesAsRead().ignore();
-      }
+    const positionsEquality = UnorderedIterableEquality<ItemPosition>();
+    if (!positionsEquality.equals(itemPositions, _markUnreadViewportSnapshot)) {
+      _markUnreadViewportDiverged = true;
     }
   }
 
+  // Marks divider A's anchor as seen once it renders on screen, or once the
+  // user scrolls past it without it ever rendering (a fast fling can skip
+  // intermediate frames). Sticky: never reverts once true, and reset only
+  // when the baseline is recaptured (channel change, or an explicit
+  // mark-unread — see [_handleCurrentUserReadChanged]).
+  //
+  // Returns true iff this call flips [_hasSeenFirstUnread] from false to
+  // true.
+  bool _maybeUpdateHasSeenFirstUnread(Iterable<ItemPosition> itemPositions) {
+    if (_isThreadConversation || _hasSeenFirstUnread.value) return false;
+
+    final anchorId = _unreadDivider.value.anchorId;
+    if (anchorId == null) return false;
+
+    final anchorMessageIndex = messages.indexWhere((it) => it.id == anchorId);
+    if (anchorMessageIndex == -1) return false;
+    final anchorItemIndex = anchorMessageIndex + 2;
+
+    final visibleIndices = itemPositions.map((position) => position.index).toList();
+    if (visibleIndices.isEmpty) return false;
+
+    final isAnchorVisible = visibleIndices.contains(anchorItemIndex);
+    // Smaller item indices are newer/closer to the bottom. If even the
+    // newest visible item is older than the anchor, the anchor has scrolled
+    // off the bottom of the viewport — the user scrolled past it.
+    final isScrolledPast = visibleIndices.reduce(min) > anchorItemIndex;
+    if (!isAnchorVisible && !isScrolledPast) return false;
+
+    _hasSeenFirstUnread.value = true;
+    return true;
+  }
+
   // Marks messages as read if the conditions are met.
   //
-  // The conditions are:
-  // 1. The channel is up to date or we are in a thread conversation.
-  // 2. There are unread messages or we are in a thread conversation.
-  // 3. In a thread, the parent has at least one reply — the server-side
-  //    thread object doesn't exist until the first reply lands.
+  // In a thread: the parent must have at least one reply (the server-side
+  // thread object doesn't exist until the first reply lands), and the
+  // channel must be up to date.
   //
-  // If any of the conditions are not met, the function returns early.
-  // Otherwise, it calls the _markMessagesAsRead function to mark the messages
-  // as read.
+  // In the channel, mirrors iOS's `shouldMarkChannelRead` gating:
+  //  1. The newest page is loaded (`isUpToDate`).
+  //  2. There is something unread to mark.
+  //  3. The bottom has been seen — either it's visible now, or it was
+  //     visible earlier and the user has since scrolled away
+  //     (`hasSeenLastMessage`).
+  //  4. If there's an active manual mark-unread (`isMarkedAsUnread`), the
+  //     viewport must genuinely differ from the one snapshotted when it
+  //     was first observed (`_markUnreadViewportSnapshot`) — otherwise the
+  //     anchor being immediately "visible" again (it's usually the very
+  //     message just marked, with nothing yet scrolled) would undo the
+  //     user's action instantly.
+  //  5. Divider A's anchor has actually been seen or scrolled past
+  //     (`hasSeenFirstUnreadMessage`) — trivially satisfied when there's
+  //     nothing to see (the channel opened fully read) or for channels using
+  //     local unread counts, mirroring iOS's escape hatch.
+  //
+  // A caller-supplied [StreamMessageListViewConfiguration.shouldMarkRead]
+  // overrides conditions 3-5.
   Future<void> _maybeMarkMessagesAsRead() async {
     final channel = streamChannel?.channel;
     if (channel == null) return;
 
     final isInThread = widget.parentMessage != null;
 
-    // A server-side thread object only exists once the parent has at least
-    // one reply; markThreadRead on a reply-less parent returns 404.
-    if (isInThread && (widget.parentMessage?.replyCount ?? 0) == 0) return;
+    if (isInThread) {
+      // A server-side thread object only exists once the parent has at
+      // least one reply; markThreadRead on a reply-less parent returns 404.
+      if ((widget.parentMessage?.replyCount ?? 0) == 0) return;
+      if (!(channel.state?.isUpToDate ?? false)) return;
+      return _debouncedMarkMessagesAsRead();
+    }
 
     final isUpToDate = channel.state?.isUpToDate ?? false;
-    if (!isInThread && !isUpToDate) return;
+    if (!isUpToDate) return;
+
+    final unreadCount = channel.state?.unreadCount ?? 0;
+    if (unreadCount <= 0) return;
+
+    final noPreexistingUnread = _unreadBaselineCaptured && _unreadBaseline == null;
+    // Equivalent to `channel.usesLocalUnreadCount`, spelled out via
+    // `channel.client` rather than `Channel`'s private client field so it
+    // stays evaluable against a test double that only implements the public
+    // API surface.
+    final usesLocalUnreadCount = channel.client.isLocalUnreadCountEnabled && !channel.canUseReadReceipts;
+    final hasSeenFirstUnreadMessage = noPreexistingUnread || _hasSeenFirstUnread.value || usesLocalUnreadCount;
+    final isMarkedAsUnread = channel.state?.isMarkedAsUnread ?? false;
+    final hasSeenLastMessage = _hasSeenLastMessage || !_showScrollToBottom.value;
+
+    if (widget.config.shouldMarkRead case final shouldMarkRead?) {
+      final details = StreamMarkReadDetails(
+        hasSeenLastMessage: hasSeenLastMessage,
+        hasSeenFirstUnreadMessage: hasSeenFirstUnreadMessage,
+        isMarkedAsUnread: isMarkedAsUnread,
+        unreadCount: unreadCount,
+      );
+      if (!shouldMarkRead(details)) return;
 
-    final hasUnread = (channel.state?.unreadCount ?? 0) > 0;
-    if (!isInThread && !hasUnread) return;
+      await _debouncedMarkMessagesAsRead();
+      _hasSeenLastMessage = false;
+      _markUnreadViewportSnapshot = null;
+      _markUnreadViewportDiverged = false;
+      return;
+    }
+
+    if (!hasSeenLastMessage) return;
+    if (!hasSeenFirstUnreadMessage) return;
+
+    if (isMarkedAsUnread) {
+      // `_handleItemPositionsChanged` already keeps this up to date on
+      // every position-changed tick; this call only matters as a fallback
+      // if this is ever reached some other way. See
+      // `_markUnreadViewportSnapshot`'s doc comment for why the guard has
+      // to latch on divergence rather than checking `isMarkedAsUnread`
+      // directly as a persistent gate.
+      _checkMarkUnreadViewportDivergence(_itemPositionListener.itemPositions.value);
+      if (!_markUnreadViewportDiverged) return;
+    }
 
-    // Mark messages as read if it's allowed.
-    return _debouncedMarkMessagesAsRead();
+    await _debouncedMarkMessagesAsRead();
+    _hasSeenLastMessage = false;
+    _markUnreadViewportSnapshot = null;
+    _markUnreadViewportDiverged = false;
   }
 
   void _getOnThreadTap() {
diff --git a/packages/stream_chat_flutter/lib/src/message_list_view/mlv_utils.dart b/packages/stream_chat_flutter/lib/src/message_list_view/mlv_utils.dart
index 5977cbefe5..c5212276ea 100644
--- a/packages/stream_chat_flutter/lib/src/message_list_view/mlv_utils.dart
+++ b/packages/stream_chat_flutter/lib/src/message_list_view/mlv_utils.dart
@@ -31,13 +31,16 @@ int getInitialIndex(
     if (targetMessageIndex != -1) return targetMessageIndex + 2;
   }
 
-  // Otherwise, return the first unread message index if available.
-  if (channelState.getFirstUnreadMessage() case final firstUnreadMessage?) {
-    final firstUnreadMessageIndex = messages.indexWhere(
-      (it) => it.id == firstUnreadMessage.id,
-    );
-
-    if (firstUnreadMessageIndex != -1) return firstUnreadMessageIndex + 2;
+  // Otherwise, return the first unread message index if available — unless
+  // the caller opted out via [StreamChannel.openAtFirstUnread].
+  if (channelState.widget.openAtFirstUnread) {
+    if (channelState.getFirstUnreadMessage() case final firstUnreadMessage?) {
+      final firstUnreadMessageIndex = messages.indexWhere(
+        (it) => it.id == firstUnreadMessage.id,
+      );
+
+      if (firstUnreadMessageIndex != -1) return firstUnreadMessageIndex + 2;
+    }
   }
 
   return 0;
diff --git a/packages/stream_chat_flutter/lib/src/message_list_view/stream_message_list_view_configuration.dart b/packages/stream_chat_flutter/lib/src/message_list_view/stream_message_list_view_configuration.dart
index 2cc256e646..263fd59bf0 100644
--- a/packages/stream_chat_flutter/lib/src/message_list_view/stream_message_list_view_configuration.dart
+++ b/packages/stream_chat_flutter/lib/src/message_list_view/stream_message_list_view_configuration.dart
@@ -1,5 +1,6 @@
 import 'package:flutter/widgets.dart';
 import 'package:stream_chat_flutter/src/message_list_view/auto_scroll_policy.dart';
+import 'package:stream_chat_flutter/src/message_list_view/mark_read_details.dart';
 
 /// {@template streamMessageListConfiguration}
 /// Holds all behavior flags and non-theme, non-builder configuration for
@@ -36,6 +37,7 @@ class StreamMessageListViewConfiguration {
     this.keyboardDismissBehavior = .onDrag,
     this.scrollPhysics = const ClampingScrollPhysics(),
     this.autoScrollPolicy = .whenOwnMessageOrAtBottom,
+    this.shouldMarkRead,
   });
 
   /// Whether to mark the channel as read when the user scrolls to the bottom.
@@ -43,6 +45,18 @@ class StreamMessageListViewConfiguration {
   /// Defaults to true.
   final bool markReadWhenAtTheBottom;
 
+  /// Overrides the built-in gating for automatic mark-read.
+  ///
+  /// When null (the default), the list marks the channel as read once the
+  /// bottom has been seen, the pre-existing unread boundary (if any) has
+  /// been seen or scrolled past, and there is no active manual mark-unread —
+  /// see [StreamMarkReadDetails]. Provide this to fully control the decision
+  /// instead.
+  ///
+  /// Only affects channel reads; has no effect on thread reads or on
+  /// [markReadWhenAtTheBottom] being `false`.
+  final StreamShouldMarkReadPredicate? shouldMarkRead;
+
   /// Whether swiping a message triggers a quoted-reply action.
   ///
   /// Defaults to false.
@@ -162,9 +176,11 @@ class StreamMessageListViewConfiguration {
     ScrollViewKeyboardDismissBehavior? keyboardDismissBehavior,
     ScrollPhysics? scrollPhysics,
     StreamAutoScrollPolicy? autoScrollPolicy,
+    StreamShouldMarkReadPredicate? shouldMarkRead,
   }) {
     return StreamMessageListViewConfiguration(
       markReadWhenAtTheBottom: markReadWhenAtTheBottom ?? this.markReadWhenAtTheBottom,
+      shouldMarkRead: shouldMarkRead ?? this.shouldMarkRead,
       swipeToReply: swipeToReply ?? this.swipeToReply,
       showScrollToBottom: showScrollToBottom ?? this.showScrollToBottom,
       showUnreadCountOnScrollToBottom: showUnreadCountOnScrollToBottom ?? this.showUnreadCountOnScrollToBottom,
@@ -204,7 +220,8 @@ class StreamMessageListViewConfiguration {
         other.retentionTrimBuffer == retentionTrimBuffer &&
         other.keyboardDismissBehavior == keyboardDismissBehavior &&
         other.scrollPhysics == scrollPhysics &&
-        other.autoScrollPolicy == autoScrollPolicy;
+        other.autoScrollPolicy == autoScrollPolicy &&
+        other.shouldMarkRead == shouldMarkRead;
   }
 
   @override
@@ -226,5 +243,6 @@ class StreamMessageListViewConfiguration {
     keyboardDismissBehavior,
     scrollPhysics,
     autoScrollPolicy,
+    shouldMarkRead,
   );
 }
diff --git a/packages/stream_chat_flutter/lib/src/message_list_view/unread_indicator_button.dart b/packages/stream_chat_flutter/lib/src/message_list_view/unread_indicator_button.dart
index 0608d19185..20727aeccc 100644
--- a/packages/stream_chat_flutter/lib/src/message_list_view/unread_indicator_button.dart
+++ b/packages/stream_chat_flutter/lib/src/message_list_view/unread_indicator_button.dart
@@ -1,15 +1,15 @@
 import 'package:flutter/material.dart';
-import 'package:stream_chat_flutter/src/misc/empty_widget.dart';
 import 'package:stream_chat_flutter/src/utils/extensions.dart';
-import 'package:stream_chat_flutter_core/stream_chat_flutter_core.dart';
 import 'package:stream_core_flutter/chat.dart' as core;
 
 /// {@template unreadIndicatorButton}
-/// A button that displays the number of unread messages in a channel.
+/// A floating "jump to unread" pill showing a fixed unread count.
 ///
-/// [UnreadIndicatorButton] listens to the current user's read state and shows
-/// a jump-to-unread button when there are unread messages. Users can tap to
-/// navigate to the oldest unread message or dismiss the indicator.
+/// [UnreadIndicatorButton] is purely presentational: the host
+/// [StreamMessageListView] decides when it should be visible (only while the
+/// pre-existing unread boundary sits above the viewport) and supplies the
+/// frozen [unreadCount]. Users can tap to navigate to the first unread
+/// message or dismiss the indicator.
 ///
 /// {@tool snippet}
 ///
@@ -17,8 +17,9 @@ import 'package:stream_core_flutter/chat.dart' as core;
 ///
 /// ```dart
 /// UnreadIndicatorButton(
-///   onJumpTap: (lastReadMessageId) async {
-///     // scroll to the unread message
+///   unreadCount: 5,
+///   onJumpTap: () async {
+///     // scroll to the first unread message
 ///   },
 ///   onDismissTap: () async {
 ///     // mark channel as read
@@ -29,21 +30,27 @@ import 'package:stream_core_flutter/chat.dart' as core;
 ///
 /// See also:
 ///
-///  * [StreamMessageListView], which hosts this widget.
+///  * [StreamMessageListView], which hosts this widget and owns its
+///    visibility.
 /// {@endtemplate}
 class UnreadIndicatorButton extends StatelessWidget {
   /// Creates an unread indicator button.
   const UnreadIndicatorButton({
     super.key,
+    required this.unreadCount,
     required this.onJumpTap,
     required this.onDismissTap,
   });
 
-  /// Called when the jump-to-unread area is tapped.
+  /// The fixed unread count to display.
   ///
-  /// Receives the ID of the last message the current user has read,
-  /// which can be used to scroll to that position.
-  final Future<void> Function(String? lastReadMessageId) onJumpTap;
+  /// This is the pre-existing unread boundary's count, captured when the
+  /// channel was opened — it does not change for the lifetime of the
+  /// session.
+  final int unreadCount;
+
+  /// Called when the jump-to-unread area is tapped.
+  final Future<void> Function() onJumpTap;
 
   /// Called when the dismiss button is tapped.
   ///
@@ -52,22 +59,10 @@ class UnreadIndicatorButton extends StatelessWidget {
 
   @override
   Widget build(BuildContext context) {
-    final channel = StreamChannel.of(context).channel;
-    if (channel.state == null) return const Empty();
-
-    return BetterStreamBuilder(
-      initialData: channel.state!.currentUserRead,
-      stream: channel.state!.currentUserReadStream,
-      builder: (context, currentUserRead) {
-        final unreadCount = currentUserRead.unreadMessages;
-        if (unreadCount <= 0) return const Empty();
-
-        return core.StreamJumpToUnreadButton(
-          label: context.translations.unreadCountIndicatorLabel(unreadCount: unreadCount),
-          onJumpPressed: () => onJumpTap(currentUserRead.lastReadMessageId),
-          onDismissPressed: onDismissTap,
-        );
-      },
+    return core.StreamJumpToUnreadButton(
+      label: context.translations.unreadCountIndicatorLabel(unreadCount: unreadCount),
+      onJumpPressed: onJumpTap,
+      onDismissPressed: onDismissTap,
     );
   }
 }
diff --git a/packages/stream_chat_flutter/lib/src/message_list_view/unread_messages_separator.dart b/packages/stream_chat_flutter/lib/src/message_list_view/unread_messages_separator.dart
index dac08c0cfe..e2480b35a5 100644
--- a/packages/stream_chat_flutter/lib/src/message_list_view/unread_messages_separator.dart
+++ b/packages/stream_chat_flutter/lib/src/message_list_view/unread_messages_separator.dart
@@ -6,8 +6,8 @@ import 'package:stream_core_flutter/chat.dart' as core;
 /// A full-width banner that marks the boundary between read and unread
 /// messages in a [StreamMessageListView].
 ///
-/// [UnreadMessagesSeparator] displays a localised "Unread Messages" label
-/// inside a subtle container with top and bottom borders.
+/// [UnreadMessagesSeparator] displays a localised "{count} unread messages"
+/// label inside a subtle container with top and bottom borders.
 ///
 /// {@tool snippet}
 ///
@@ -106,7 +106,7 @@ class UnreadMessagesSeparator extends StatelessWidget {
         child: Padding(
           padding: effectiveContentPadding,
           child: Text(
-            context.translations.unreadMessagesSeparatorText(),
+            context.translations.unreadMessagesSeparatorLabel(count: unreadCount),
             textAlign: TextAlign.center,
             style: effectiveTextStyle,
           ),
diff --git a/packages/stream_chat_flutter/lib/stream_chat_flutter.dart b/packages/stream_chat_flutter/lib/stream_chat_flutter.dart
index 6a1c377bcd..be5848144f 100644
--- a/packages/stream_chat_flutter/lib/stream_chat_flutter.dart
+++ b/packages/stream_chat_flutter/lib/stream_chat_flutter.dart
@@ -86,10 +86,12 @@ export 'src/message_input/stream_message_composer.dart';
 export 'src/message_input/stream_message_composer_attachment_list.dart';
 export 'src/message_input/stream_message_text_field.dart';
 export 'src/message_list_view/auto_scroll_policy.dart';
+export 'src/message_list_view/mark_read_details.dart';
 export 'src/message_list_view/message_list_view.dart';
 export 'src/message_list_view/stream_message_list_view_builders.dart';
 export 'src/message_list_view/stream_message_list_view_configuration.dart';
 export 'src/message_list_view/unread_indicator_button.dart';
+export 'src/message_list_view/unread_messages_separator.dart';
 export 'src/message_modal/message_action_confirmation_modal.dart';
 export 'src/message_modal/message_actions_modal.dart';
 export 'src/message_modal/message_modal.dart';
diff --git a/packages/stream_chat_flutter/test/src/localization/default_translations_test.dart b/packages/stream_chat_flutter/test/src/localization/default_translations_test.dart
index 0950c6832e..22c90669fd 100644
--- a/packages/stream_chat_flutter/test/src/localization/default_translations_test.dart
+++ b/packages/stream_chat_flutter/test/src/localization/default_translations_test.dart
@@ -183,6 +183,8 @@ void main() {
     expect(translations.replyToMessageLabel, isNotNull);
     expect(translations.unreadCountIndicatorLabel(unreadCount: 2), isNotNull);
     expect(translations.unreadMessagesSeparatorText(), isNotNull);
+    expect(translations.unreadMessagesSeparatorLabel(count: 1), '1 unread message');
+    expect(translations.unreadMessagesSeparatorLabel(count: 2), '2 unread messages');
     expect(translations.markUnreadError, isNotNull);
     expect(translations.markAsUnreadLabel, isNotNull);
     expect(translations.toggleBlockUnblockUserText(isBlocked: false), isNotNull);
diff --git a/packages/stream_chat_flutter/test/src/message_list_view/mark_read_test.dart b/packages/stream_chat_flutter/test/src/message_list_view/mark_read_test.dart
index 3d2191e90f..27efe5dda3 100644
--- a/packages/stream_chat_flutter/test/src/message_list_view/mark_read_test.dart
+++ b/packages/stream_chat_flutter/test/src/message_list_view/mark_read_test.dart
@@ -1,16 +1,22 @@
 // Tests for `StreamMessageListView`'s mark-read-at-the-bottom behavior.
 //
-// The logic lives in `_handleLastItemFullyVisible` →
-// `_maybeMarkMessagesAsRead`. It fires `channel.markRead()` when the user
-// reaches the bottom of the list, gated on:
+// The logic lives in `_handleItemPositionsChanged` →
+// `_maybeMarkMessagesAsRead` (FLU-640). Marking the channel read requires all
+// of:
 //
 //   1. `markReadWhenAtTheBottom` is true (the default).
 //   2. `channel.state.isUpToDate` is true (or we're in a thread).
 //   3. `channel.state.unreadCount > 0`.
+//   4. The bottom has been seen (now, or earlier then scrolled away).
+//   5. The pre-existing unread boundary (if any) has been seen or scrolled
+//      past — trivially satisfied when the channel opened fully read.
+//   6. There is no active manual mark-unread (`channel.state.isMarkedAsUnread`).
 //
-// In a thread, it fires `channel.markThreadRead(parentId)` instead, and is
-// additionally gated on the parent having at least one reply — the server-side
-// thread object only exists after the first reply, so an earlier call 404s.
+// `StreamMessageListViewConfiguration.shouldMarkRead` can override 4-6.
+//
+// In a thread, it fires `channel.markThreadRead(parentId)` instead, gated
+// only on the parent having at least one reply and the channel being up to
+// date — conditions 4-6 don't apply there.
 //
 // These tests pin the expected behavior so regressions in the underlying
 // position-listener flow (SPL `itemPositions`, scroll wiring, etc.) surface
@@ -32,11 +38,13 @@ void main() {
   late Channel channel;
   late ChannelClientState channelClientState;
   late ClientState clientState;
+  late OwnUser ownUser;
 
   late StreamController<bool> isUpToDateController;
   late StreamController<int> unreadCountController;
   late StreamController<List<Message>> messagesController;
   late StreamController<Map<String, List<Message>>> threadsController;
+  late StreamController<Read?> currentUserReadController;
 
   setUpAll(() {
     registerFallbackValue(EventType.messageNew);
@@ -46,23 +54,26 @@ void main() {
     client = MockClient();
     clientState = MockClientState();
     when(() => client.state).thenAnswer((_) => clientState);
-    final own = OwnUser(id: 'ownid');
-    when(() => clientState.currentUser).thenReturn(own);
-    when(() => clientState.currentUserStream).thenAnswer((_) => Stream.value(own));
+    ownUser = OwnUser(id: 'ownid');
+    when(() => clientState.currentUser).thenReturn(ownUser);
+    when(() => clientState.currentUserStream).thenAnswer((_) => Stream.value(ownUser));
 
     channel = MockChannel();
     channelClientState = MockChannelState();
     when(() => channel.client).thenReturn(client);
     when(() => channel.state).thenReturn(channelClientState);
+    when(() => client.isLocalUnreadCountEnabled).thenReturn(false);
 
     isUpToDateController = StreamController<bool>.broadcast();
     unreadCountController = StreamController<int>.broadcast();
     messagesController = StreamController<List<Message>>.broadcast();
     threadsController = StreamController<Map<String, List<Message>>>.broadcast();
+    currentUserReadController = StreamController<Read?>.broadcast();
     addTearDown(isUpToDateController.close);
     addTearDown(unreadCountController.close);
     addTearDown(messagesController.close);
     addTearDown(threadsController.close);
+    addTearDown(currentUserReadController.close);
 
     when(() => channelClientState.threadsStream).thenAnswer((_) => threadsController.stream);
     when(() => channelClientState.threads).thenReturn(const {});
@@ -72,9 +83,9 @@ void main() {
     when(() => channelClientState.read).thenReturn([]);
     when(() => channelClientState.membersStream).thenAnswer((_) => const Stream.empty());
     when(() => channelClientState.members).thenReturn([]);
-    when(() => channelClientState.currentUserRead).thenReturn(null);
-    when(() => channelClientState.currentUserReadStream).thenAnswer((_) => const Stream.empty());
+    when(() => channelClientState.currentUserReadStream).thenAnswer((_) => currentUserReadController.stream);
     when(() => channelClientState.messagesStream).thenAnswer((_) => messagesController.stream);
+    when(() => channelClientState.isMarkedAsUnread).thenReturn(false);
 
     // Mark-read mocks return immediately.
     when(() => channel.markRead(messageId: any(named: 'messageId'))).thenAnswer((_) async => EmptyResponse());
@@ -89,6 +100,12 @@ void main() {
     ).thenAnswer((_) async => QueryRepliesResponse()..messages = []);
   });
 
+  // Default: opened with nothing pre-existing unread, so the FLU-640
+  // "has seen the first unread boundary" condition is trivially satisfied
+  // and doesn't gate these tests unless a `currentUserRead` override says
+  // otherwise.
+  Read noPreexistingUnreadRead() => Read(user: ownUser, lastRead: DateTime.now(), unreadMessages: 0);
+
   Future<void> pumpMessageList(
     WidgetTester tester, {
     required List<Message> messages,
@@ -96,11 +113,17 @@ void main() {
     required int unreadCount,
     bool markReadWhenAtTheBottom = true,
     Message? parentMessage,
+    Read? currentUserRead,
+    bool openAtFirstUnread = false,
+    StreamShouldMarkReadPredicate? shouldMarkRead,
   }) async {
     when(() => channelClientState.isUpToDate).thenReturn(isUpToDate);
     when(() => channelClientState.unreadCount).thenReturn(unreadCount);
     when(() => channelClientState.messages).thenReturn(messages);
 
+    final resolvedRead = currentUserRead ?? noPreexistingUnreadRead();
+    when(() => channelClientState.currentUserRead).thenReturn(resolvedRead);
+
     // In thread mode, MessageListCore reads from state.threads[parentId] and
     // subscribes to state.threadsStream. Seed both so the reply list renders.
     if (parentMessage != null) {
@@ -117,10 +140,12 @@ void main() {
               themeData: StreamChatThemeData(),
               child: StreamChannel(
                 channel: channel,
+                openAtFirstUnread: openAtFirstUnread,
                 child: StreamMessageListView(
                   parentMessage: parentMessage,
                   config: StreamMessageListViewConfiguration(
                     markReadWhenAtTheBottom: markReadWhenAtTheBottom,
+                    shouldMarkRead: shouldMarkRead,
                   ),
                 ),
               ),
@@ -131,6 +156,7 @@ void main() {
       // Prime the streams.
       isUpToDateController.add(isUpToDate);
       unreadCountController.add(unreadCount);
+      currentUserReadController.add(resolvedRead);
       if (parentMessage != null) {
         threadsController.add({parentMessage.id: messages});
       } else {
@@ -146,7 +172,7 @@ void main() {
       'unreadCount>0',
       (tester) async {
         final other = User(id: 'otherid');
-        final messages = generateConversation(20, users: [other]);
+        final messages = generateConversation(20, users: [other]).reversed.toList();
 
         await pumpMessageList(
           tester,
@@ -163,7 +189,7 @@ void main() {
       'does NOT fire when isUpToDate=false (gate on incomplete state)',
       (tester) async {
         final other = User(id: 'otherid');
-        final messages = generateConversation(20, users: [other]);
+        final messages = generateConversation(20, users: [other]).reversed.toList();
 
         await pumpMessageList(
           tester,
@@ -182,7 +208,7 @@ void main() {
       'does NOT fire when unreadCount is 0',
       (tester) async {
         final other = User(id: 'otherid');
-        final messages = generateConversation(20, users: [other]);
+        final messages = generateConversation(20, users: [other]).reversed.toList();
 
         await pumpMessageList(
           tester,
@@ -201,7 +227,7 @@ void main() {
       'does NOT fire when markReadWhenAtTheBottom is false',
       (tester) async {
         final other = User(id: 'otherid');
-        final messages = generateConversation(20, users: [other]);
+        final messages = generateConversation(20, users: [other]).reversed.toList();
 
         await pumpMessageList(
           tester,
@@ -216,6 +242,132 @@ void main() {
         );
       },
     );
+
+    testWidgets(
+      'does NOT fire when opened at the bottom with an unseen pre-existing '
+      'unread boundary (FLU-640)',
+      (tester) async {
+        final other = User(id: 'otherid');
+        final messages = generateConversation(20, users: [other]).reversed.toList();
+        // A boundary partway through the list — the user hasn't scrolled up
+        // to see it since the list opens at the bottom.
+        final lastReadMessageId = messages[10].id;
+
+        await pumpMessageList(
+          tester,
+          messages: messages,
+          isUpToDate: true,
+          unreadCount: 5,
+          openAtFirstUnread: false,
+          currentUserRead: Read(
+            user: ownUser,
+            lastRead: DateTime.now(),
+            unreadMessages: 5,
+            lastReadMessageId: lastReadMessageId,
+          ),
+        );
+
+        verifyNever(() => channel.markRead(messageId: any(named: 'messageId')));
+      },
+    );
+
+    testWidgets(
+      'does NOT fire when the channel has an active manual mark-unread',
+      (tester) async {
+        final other = User(id: 'otherid');
+        final messages = generateConversation(20, users: [other]).reversed.toList();
+        when(() => channelClientState.isMarkedAsUnread).thenReturn(true);
+
+        await pumpMessageList(
+          tester,
+          messages: messages,
+          isUpToDate: true,
+          unreadCount: 5,
+        );
+
+        verifyNever(() => channel.markRead(messageId: any(named: 'messageId')));
+      },
+    );
+
+    testWidgets(
+      'fires once the viewport genuinely changes after mounting with an '
+      'already-active manual mark-unread (no live transition for '
+      '_handleCurrentUserReadChanged to hook the snapshot on)',
+      (tester) async {
+        final other = User(id: 'otherid');
+        final messages = generateConversation(40, users: [other]).reversed.toList();
+        when(() => channelClientState.isMarkedAsUnread).thenReturn(true);
+
+        await pumpMessageList(
+          tester,
+          messages: messages,
+          isUpToDate: true,
+          unreadCount: 5,
+        );
+
+        // The very first layout — at the bottom, nothing scrolled yet — is
+        // exactly the moment the viewport snapshot gets captured. Marking
+        // read here would be the reintroduced deadlock: a channel that
+        // opens already marked unread and happens to land at the bottom
+        // would get instantly marked read again before the user did
+        // anything.
+        verifyNever(() => channel.markRead(messageId: any(named: 'messageId')));
+
+        // A genuine scroll away and back changes the viewport, proving the
+        // user did something since the snapshot was taken — this should
+        // no longer be blocked.
+        await tester.drag(find.byType(StreamMessageListView), const Offset(0, 400));
+        await tester.pumpAndSettle();
+        await tester.drag(find.byType(StreamMessageListView), const Offset(0, -1000));
+        await tester.pumpAndSettle();
+
+        verify(() => channel.markRead(messageId: any(named: 'messageId'))).called(1);
+      },
+    );
+
+    testWidgets(
+      'a shouldMarkRead override that returns false blocks an otherwise-allowed mark-read',
+      (tester) async {
+        final other = User(id: 'otherid');
+        final messages = generateConversation(20, users: [other]).reversed.toList();
+
+        await pumpMessageList(
+          tester,
+          messages: messages,
+          isUpToDate: true,
+          unreadCount: 5,
+          shouldMarkRead: (details) => false,
+        );
+
+        verifyNever(() => channel.markRead(messageId: any(named: 'messageId')));
+      },
+    );
+
+    testWidgets(
+      'a shouldMarkRead override that returns true allows a mark-read the default gating would block',
+      (tester) async {
+        final other = User(id: 'otherid');
+        final messages = generateConversation(20, users: [other]).reversed.toList();
+        final lastReadMessageId = messages[10].id;
+
+        await pumpMessageList(
+          tester,
+          messages: messages,
+          isUpToDate: true,
+          unreadCount: 5,
+          openAtFirstUnread: false,
+          currentUserRead: Read(
+            user: ownUser,
+            lastRead: DateTime.now(),
+            unreadMessages: 5,
+            lastReadMessageId: lastReadMessageId,
+          ),
+          shouldMarkRead: (details) => true,
+        );
+
+        verify(() => channel.markRead(messageId: any(named: 'messageId'))).called(1);
+      },
+    );
   });
 
   group('thread markThreadRead gates', () {
@@ -310,7 +462,7 @@ void main() {
       'is hidden when the user lands at the bottom with isUpToDate=true',
       (tester) async {
         final other = User(id: 'otherid');
-        final messages = generateConversation(20, users: [other]);
+        final messages = generateConversation(20, users: [other]).reversed.toList();
 
         await pumpMessageList(
           tester,
@@ -319,8 +471,134 @@ void main() {
           unreadCount: 0,
         );
 
-        // The default scroll-to-bottom button is a FloatingActionButton.
-        expect(find.byType(FloatingActionButton), findsNothing);
+        // The default scroll-to-bottom button is a floating StreamButton,
+        // shown only while scrolled away from the bottom.
+        expect(find.byType(StreamButton), findsNothing);
+      },
+    );
+  });
+
+  group('unread pill', () {
+    testWidgets(
+      'shown when opened at the bottom with an unseen pre-existing unread boundary',
+      (tester) async {
+        final other = User(id: 'otherid');
+        final messages = generateConversation(20, users: [other]).reversed.toList();
+        final lastReadMessageId = messages[10].id;
+
+        await pumpMessageList(
+          tester,
+          messages: messages,
+          isUpToDate: true,
+          unreadCount: 5,
+          openAtFirstUnread: false,
+          currentUserRead: Read(
+            user: ownUser,
+            lastRead: DateTime.now(),
+            unreadMessages: 5,
+            lastReadMessageId: lastReadMessageId,
+          ),
+        );
+
+        expect(find.byType(UnreadIndicatorButton), findsOneWidget);
+      },
+    );
+
+    testWidgets(
+      'absent when the channel opened with nothing pre-existing unread',
+      (tester) async {
+        final other = User(id: 'otherid');
+        final messages = generateConversation(20, users: [other]).reversed.toList();
+
+        await pumpMessageList(
+          tester,
+          messages: messages,
+          isUpToDate: true,
+          unreadCount: 0,
+          openAtFirstUnread: false,
+        );
+
+        expect(find.byType(UnreadIndicatorButton), findsNothing);
+      },
+    );
+
+    testWidgets(
+      'is shown immediately even when the boundary message has not loaded yet '
+      '(top pagination pending)',
+      (tester) async {
+        final other = User(id: 'otherid');
+        final messages = generateConversation(20, users: [other]).reversed.toList();
+        // Not part of the loaded window — simulates the read boundary sitting
+        // further back in history than top pagination has reached yet. The
+        // pill's count is known from the `Read` itself, so it shouldn't have
+        // to wait on the anchor message to load before appearing.
+        const lastReadMessageId = 'not-yet-loaded-message-id';
+
+        await pumpMessageList(
+          tester,
+          messages: messages,
+          isUpToDate: true,
+          unreadCount: 5,
+          openAtFirstUnread: false,
+          currentUserRead: Read(
+            user: ownUser,
+            lastRead: DateTime.now(),
+            unreadMessages: 5,
+            lastReadMessageId: lastReadMessageId,
+          ),
+        );
+
+        final indicator = tester.widget<UnreadIndicatorButton>(
+          find.byType(UnreadIndicatorButton),
+        );
+        expect(indicator.unreadCount, 5);
+      },
+    );
+
+    testWidgets(
+      "tapping jump before the anchor resolves falls back to the boundary's "
+      'lastReadMessageId instead of doing nothing',
+      (tester) async {
+        final other = User(id: 'otherid');
+        final messages = generateConversation(20, users: [other]).reversed.toList();
+        const lastReadMessageId = 'not-yet-loaded-message-id';
+
+        when(
+          () => channel.query(
+            preferOffline: any(named: 'preferOffline'),
+            messagesPagination: any(named: 'messagesPagination'),
+          ),
+        ).thenAnswer((_) async => const ChannelState(messages: []));
+
+        await pumpMessageList(
+          tester,
+          messages: messages,
+          isUpToDate: true,
+          unreadCount: 5,
+          openAtFirstUnread: false,
+          currentUserRead: Read(
+            user: ownUser,
+            lastRead: DateTime.now(),
+            unreadMessages: 5,
+            lastReadMessageId: lastReadMessageId,
+          ),
+        );
+
+        final indicator = tester.widget<UnreadIndicatorButton>(
+          find.byType(UnreadIndicatorButton),
+        );
+        // Not awaited directly: `_scrollToMessage`'s fallback awaits
+        // `WidgetsBinding.instance.endOfFrame` after the query, which only
+        // resolves once the test binding actually pumps a frame.
+        unawaited(indicator.onJumpTap());
+        await tester.pumpAndSettle();
+
+        verify(
+          () => channel.query(
+            preferOffline: false,
+            messagesPagination: const PaginationParams(limit: 30, idAround: lastReadMessageId),
+          ),
+        ).called(1);
       },
     );
   });
@@ -330,7 +608,8 @@ void main() {
       'marks the channel read immediately when tapped',
       (tester) async {
         final other = User(id: 'otherid');
-        final messages = generateConversation(20, users: [other]);
+        final messages = generateConversation(20, users: [other]).reversed.toList();
+        final lastReadMessageId = messages[10].id;
 
         await pumpMessageList(
           tester,
@@ -338,6 +617,13 @@ void main() {
           isUpToDate: true,
           unreadCount: 5,
           markReadWhenAtTheBottom: false,
+          openAtFirstUnread: false,
+          currentUserRead: Read(
+            user: ownUser,
+            lastRead: DateTime.now(),
+            unreadMessages: 5,
+            lastReadMessageId: lastReadMessageId,
+          ),
         );
 
         // Nothing has marked the channel read yet.
@@ -356,7 +642,8 @@ void main() {
       'fires markRead on every tap (not debounced)',
       (tester) async {
         final other = User(id: 'otherid');
-        final messages = generateConversation(20, users: [other]);
+        final messages = generateConversation(20, users: [other]).reversed.toList();
+        final lastReadMessageId = messages[10].id;
 
         await pumpMessageList(
           tester,
@@ -364,6 +651,13 @@ void main() {
           isUpToDate: true,
           unreadCount: 5,
           markReadWhenAtTheBottom: false,
+          openAtFirstUnread: false,
+          currentUserRead: Read(
+            user: ownUser,
+            lastRead: DateTime.now(),
+            unreadMessages: 5,
+            lastReadMessageId: lastReadMessageId,
+          ),
         );
 
         final indicator = tester.widget<UnreadIndicatorButton>(
@@ -380,5 +674,37 @@ void main() {
         verify(() => channel.markRead(messageId: any(named: 'messageId'))).called(3);
       },
     );
+
+    testWidgets(
+      'dismisses the pill permanently, even though markReadWhenAtTheBottom is off',
+      (tester) async {
+        final other = User(id: 'otherid');
+        final messages = generateConversation(20, users: [other]).reversed.toList();
+        final lastReadMessageId = messages[10].id;
+
+        await pumpMessageList(
+          tester,
+          messages: messages,
+          isUpToDate: true,
+          unreadCount: 5,
+          markReadWhenAtTheBottom: false,
+          openAtFirstUnread: false,
+          currentUserRead: Read(
+            user: ownUser,
+            lastRead: DateTime.now(),
+            unreadMessages: 5,
+            lastReadMessageId: lastReadMessageId,
+          ),
+        );
+
+        final indicator = tester.widget<UnreadIndicatorButton>(
+          find.byType(UnreadIndicatorButton),
+        );
+        await indicator.onDismissTap();
+        await tester.pumpAndSettle();
+
+        expect(find.byType(UnreadIndicatorButton), findsNothing);
+      },
+    );
   });
 }
diff --git a/packages/stream_chat_flutter/test/src/message_list_view/unread_divider_test.dart b/packages/stream_chat_flutter/test/src/message_list_view/unread_divider_test.dart
new file mode 100644
index 0000000000..e431d94d8f
--- /dev/null
+++ b/packages/stream_chat_flutter/test/src/message_list_view/unread_divider_test.dart
@@ -0,0 +1,429 @@
+// Tests for the unread-messages divider, the jump-to-unread pill, and the
+// scroll-to-bottom badge (FLU-649 / FLU-650).
+//
+//  - The unread divider ("{n} unread messages"): anchored to the
+//    pre-existing unread boundary captured when the channel opens. The
+//    anchor is frozen — it stays on screen for the whole session regardless
+//    of scrolling or reads — but its displayed count keeps counting up as
+//    further messages arrive out of view during the session, mirroring
+//    WhatsApp, rather than staying frozen at the open-time count.
+//  - The pill shows the count of unread messages captured when the channel
+//    was opened — this one *does* stay frozen — and is gated on that
+//    boundary being above the viewport.
+//  - The scroll-to-bottom badge counts messages that arrive while the user
+//    is scrolled away from the bottom, and always resets to 0 once they
+//    reach the bottom.
+
+import 'dart:async';
+
+import 'package:flutter/material.dart';
+import 'package:flutter/services.dart';
+import 'package:flutter_test/flutter_test.dart';
+import 'package:mocktail/mocktail.dart';
+import 'package:stream_chat_flutter/stream_chat_flutter.dart';
+
+import '../../test_utils/data_generator.dart';
+import '../mocks.dart';
+
+void main() {
+  late StreamChatClient client;
+  late Channel channel;
+  late ChannelClientState channelClientState;
+  late ClientState clientState;
+  late OwnUser ownUser;
+
+  late StreamController<bool> isUpToDateController;
+  late StreamController<int> unreadCountController;
+  late StreamController<List<Message>> messagesController;
+  late StreamController<Event> messageNewController;
+  late StreamController<Read?> currentUserReadController;
+
+  setUpAll(() {
+    registerFallbackValue(EventType.messageNew);
+  });
+
+  setUp(() {
+    client = MockClient();
+    clientState = MockClientState();
+    when(() => client.state).thenAnswer((_) => clientState);
+    ownUser = OwnUser(id: 'ownid');
+    when(() => clientState.currentUser).thenReturn(ownUser);
+    when(() => clientState.currentUserStream).thenAnswer((_) => Stream.value(ownUser));
+    when(() => client.isLocalUnreadCountEnabled).thenReturn(false);
+
+    isUpToDateController = StreamController<bool>.broadcast();
+    unreadCountController = StreamController<int>.broadcast();
+    messagesController = StreamController<List<Message>>.broadcast();
+    // MockChannel.on filters this by event.type, so events pushed here
+    // surface to channel.on(EventType.messageNew) subscribers.
+    messageNewController = StreamController<Event>.broadcast();
+    currentUserReadController = StreamController<Read?>.broadcast();
+    addTearDown(isUpToDateController.close);
+    addTearDown(unreadCountController.close);
+    addTearDown(messagesController.close);
+    addTearDown(messageNewController.close);
+    addTearDown(currentUserReadController.close);
+
+    channel = MockChannel(eventStream: messageNewController.stream);
+    channelClientState = MockChannelState();
+    when(() => channel.client).thenReturn(client);
+    when(() => channel.state).thenReturn(channelClientState);
+
+    when(() => channelClientState.threadsStream).thenAnswer((_) => const Stream.empty());
+    when(() => channelClientState.isUpToDateStream).thenAnswer((_) => isUpToDateController.stream);
+    when(() => channelClientState.unreadCountStream).thenAnswer((_) => unreadCountController.stream);
+    when(() => channelClientState.readStream).thenAnswer((_) => const Stream.empty());
+    when(() => channelClientState.read).thenReturn([]);
+    when(() => channelClientState.membersStream).thenAnswer((_) => const Stream.empty());
+    when(() => channelClientState.members).thenReturn([]);
+    when(() => channelClientState.currentUserReadStream).thenAnswer((_) => currentUserReadController.stream);
+    when(() => channelClientState.messagesStream).thenAnswer((_) => messagesController.stream);
+    when(() => channelClientState.isMarkedAsUnread).thenReturn(false);
+
+    when(() => channel.markRead(messageId: any(named: 'messageId'))).thenAnswer((_) async => EmptyResponse());
+  });
+
+  Future<void> pumpMessageList(
+    WidgetTester tester, {
+    required List<Message> messages,
+    bool isUpToDate = true,
+    required int unreadCount,
+    required Read currentUserRead,
+    bool openAtFirstUnread = false,
+  }) async {
+    when(() => channelClientState.isUpToDate).thenReturn(isUpToDate);
+    when(() => channelClientState.unreadCount).thenReturn(unreadCount);
+    when(() => channelClientState.messages).thenReturn(messages);
+    when(() => channelClientState.currentUserRead).thenReturn(currentUserRead);
+
+    await tester.runAsync(() async {
+      await tester.pumpWidget(
+        MaterialApp(
+          home: DefaultAssetBundle(
+            bundle: rootBundle,
+            child: StreamChat(
+              client: client,
+              themeData: StreamChatThemeData(),
+              child: StreamChannel(
+                channel: channel,
+                openAtFirstUnread: openAtFirstUnread,
+                child: const StreamMessageListView(
+                  config: StreamMessageListViewConfiguration(
+                    markReadWhenAtTheBottom: false,
+                    // Own messages otherwise auto-scroll back to the
+                    // bottom by default, which would confound these
+                    // tests' control over scroll position.
+                    autoScrollPolicy: StreamAutoScrollPolicy.disabled,
+                  ),
+                ),
+              ),
+            ),
+          ),
+        ),
+      );
+      isUpToDateController.add(isUpToDate);
+      unreadCountController.add(unreadCount);
+      currentUserReadController.add(currentUserRead);
+      messagesController.add(messages);
+      await tester.pumpAndSettle();
+    });
+  }
+
+  // Appends to the end because production state.messages is oldest-first.
+  Future<void> deliverMessageNew(
+    WidgetTester tester, {
+    required Message newMessage,
+    required List<Message> existing,
+  }) async {
+    final updated = [...existing, newMessage];
+    when(() => channelClientState.messages).thenReturn(updated);
+    await tester.runAsync(() async {
+      messagesController.add(updated);
+      messageNewController.add(Event(type: EventType.messageNew, message: newMessage, cid: channel.cid));
+      await tester.pumpAndSettle();
+    });
+  }
+
+  group('unread divider (pre-existing unread)', () {
+    testWidgets(
+      'shows the open-time count and stays visible after unreadCount drops to 0',
+      (tester) async {
+        final other = User(id: 'otherid');
+        final messages = generateConversation(20, users: [other]).reversed.toList();
+        // Close to the bottom so the anchor message — and its divider — are
+        // guaranteed to be within the initially-rendered window regardless
+        // of viewport size; the list opens at the bottom (openAtFirstUnread
+        // is false in this helper) and SPL only builds visible items.
+        final lastReadMessageId = messages[messages.length - 3].id;
+
+        await pumpMessageList(
+          tester,
+          messages: messages,
+          unreadCount: 2,
+          currentUserRead: Read(
+            user: ownUser,
+            lastRead: DateTime.now(),
+            unreadMessages: 2,
+            lastReadMessageId: lastReadMessageId,
+          ),
+        );
+
+        expect(find.text('2 unread messages'), findsOneWidget);
+
+        // Simulate an auto mark-read completing server-side: the live count
+        // drops to 0, but the divider must not react to it — its anchor and
+        // open-time count are frozen.
+        unreadCountController.add(0);
+        when(() => channelClientState.unreadCount).thenReturn(0);
+        await tester.pumpAndSettle();
+
+        expect(find.text('2 unread messages'), findsOneWidget);
+      },
+    );
+
+    testWidgets(
+      'is absent when the channel opened with nothing pre-existing unread',
+      (tester) async {
+        final other = User(id: 'otherid');
+        final messages = generateConversation(20, users: [other]).reversed.toList();
+
+        await pumpMessageList(
+          tester,
+          messages: messages,
+          unreadCount: 0,
+          currentUserRead: Read(user: ownUser, lastRead: DateTime.now(), unreadMessages: 0),
+        );
+
+        expect(find.textContaining('unread message'), findsNothing);
+      },
+    );
+
+    testWidgets(
+      'keeps counting up as further messages arrive out of view',
+      (tester) async {
+        final other = User(id: 'otherid');
+        final messages = generateConversation(20, users: [other]).reversed.toList();
+        final lastReadMessageId = messages[messages.length - 3].id;
+
+        await pumpMessageList(
+          tester,
+          messages: messages,
+          unreadCount: 2,
+          currentUserRead: Read(
+            user: ownUser,
+            lastRead: DateTime.now(),
+            unreadMessages: 2,
+            lastReadMessageId: lastReadMessageId,
+          ),
+        );
+
+        expect(find.text('2 unread messages'), findsOneWidget);
+
+        // Scroll just enough away from the bottom (to flip `isAtBottom`)
+        // while keeping the anchor, close to the bottom, within SPL's
+        // rendered window — an out-of-view arrival should then grow the
+        // divider's count on top of the open-time baseline.
+        await tester.drag(find.byType(StreamMessageListView), const Offset(0, 120));
+        await tester.pumpAndSettle();
+
+        final fromOther = Message(
+          id: 'new-from-other-growth-probe',
+          text: 'Out of view',
+          user: other,
+          createdAt: DateTime.now(),
+        );
+        await deliverMessageNew(tester, newMessage: fromOther, existing: messages);
+
+        expect(find.text('3 unread messages'), findsOneWidget);
+
+        // A second out-of-view arrival grows it further.
+        final secondFromOther = Message(
+          id: 'second-new-from-other-growth-probe',
+          text: 'Also out of view',
+          user: other,
+          createdAt: DateTime.now(),
+        );
+        await deliverMessageNew(tester, newMessage: secondFromOther, existing: [...messages, fromOther]);
+
+        expect(find.text('4 unread messages'), findsOneWidget);
+      },
+    );
+
+    testWidgets(
+      'also grows for arrivals seen live at the bottom, unlike the badge',
+      (tester) async {
+        final other = User(id: 'otherid');
+        final messages = generateConversation(20, users: [other]).reversed.toList();
+        final lastReadMessageId = messages[messages.length - 3].id;
+
+        await pumpMessageList(
+          tester,
+          messages: messages,
+          unreadCount: 2,
+          currentUserRead: Read(
+            user: ownUser,
+            lastRead: DateTime.now(),
+            unreadMessages: 2,
+            lastReadMessageId: lastReadMessageId,
+          ),
+        );
+
+        expect(find.text('2 unread messages'), findsOneWidget);
+
+        // Still at the bottom — no scrolling away — an arrival here would
+        // never bump the scroll-to-bottom badge, but the divider isn't
+        // "caught up" the way the badge's out-of-view count is; it should
+        // keep counting every arrival regardless of scroll position.
+        final fromOther = Message(
+          id: 'new-from-other-at-bottom-probe',
+          text: 'Seen immediately',
+          user: other,
+          createdAt: DateTime.now(),
+        );
+        await deliverMessageNew(tester, newMessage: fromOther, existing: messages);
+
+        expect(find.text('3 unread messages'), findsOneWidget);
+      },
+    );
+  });
+
+  // The badge is a floating overlay, always built regardless of scroll
+  // position — unlike the inline divider, which only exists in the widget
+  // tree once SPL actually renders its anchor message.
+  String? badgeLabel(WidgetTester tester) {
+    final finder = find.byType(StreamBadgeNotification);
+    if (finder.evaluate().isEmpty) return null;
+    return tester.widget<StreamBadgeNotification>(finder).props.label;
+  }
+
+  group('scroll-to-bottom badge', () {
+    testWidgets(
+      'appears only once the user is scrolled away from the bottom, and skips own messages',
+      (tester) async {
+        final other = User(id: 'otherid');
+        final messages = generateConversation(40, users: [other]).reversed.toList();
+
+        await pumpMessageList(
+          tester,
+          messages: messages,
+          unreadCount: 0,
+          currentUserRead: Read(user: ownUser, lastRead: DateTime.now(), unreadMessages: 0),
+        );
+
+        // At the bottom: an arrival is in view, so it shouldn't bump the
+        // badge.
+        final whileAtBottom = Message(
+          id: 'while-at-bottom',
+          text: 'Seen immediately',
+          user: other,
+          createdAt: DateTime.now(),
+        );
+        await deliverMessageNew(tester, newMessage: whileAtBottom, existing: messages);
+
+        expect(badgeLabel(tester), isNull);
+
+        // Scroll away from the bottom, then a message from another user
+        // should bump the badge.
+        await tester.drag(find.byType(StreamMessageListView), const Offset(0, 400));
+        await tester.pumpAndSettle();
+
+        final fromOther = Message(
+          id: 'new-from-other',
+          text: 'Out of view',
+          user: other,
+          createdAt: DateTime.now(),
+        );
+        await deliverMessageNew(tester, newMessage: fromOther, existing: [...messages, whileAtBottom]);
+
+        expect(badgeLabel(tester), '1');
+
+        // A second out-of-view arrival grows the count.
+        final secondFromOther = Message(
+          id: 'second-new-from-other',
+          text: 'Also out of view',
+          user: other,
+          createdAt: DateTime.now(),
+        );
+        await deliverMessageNew(tester, newMessage: secondFromOther, existing: [...messages, whileAtBottom, fromOther]);
+
+        expect(badgeLabel(tester), '2');
+      },
+    );
+
+    testWidgets(
+      "the current user's own messages don't count towards the badge",
+      (tester) async {
+        final other = User(id: 'otherid');
+        final messages = generateConversation(40, users: [other]).reversed.toList();
+
+        await pumpMessageList(
+          tester,
+          messages: messages,
+          unreadCount: 0,
+          currentUserRead: Read(user: ownUser, lastRead: DateTime.now(), unreadMessages: 0),
+        );
+
+        await tester.drag(find.byType(StreamMessageListView), const Offset(0, 400));
+        await tester.pumpAndSettle();
+
+        // Auto-scroll is disabled in this helper's config, so an own
+        // message while scrolled up genuinely stays out of view too — this
+        // isolates "does it count" from "does it pull me back to the
+        // bottom" (a separate, already-covered concern in auto_scroll_test).
+        final ownMessage = Message(
+          id: 'own-while-scrolled-up',
+          text: 'My own message',
+          user: ownUser,
+          createdAt: DateTime.now(),
+        );
+        await deliverMessageNew(tester, newMessage: ownMessage, existing: messages);
+
+        expect(badgeLabel(tester), isNull);
+      },
+    );
+
+    testWidgets(
+      'always resets to 0 once the user reaches the bottom',
+      (tester) async {
+        final other = User(id: 'otherid');
+        final messages = generateConversation(40, users: [other]).reversed.toList();
+
+        await pumpMessageList(
+          tester,
+          messages: messages,
+          unreadCount: 0,
+          currentUserRead: Read(user: ownUser, lastRead: DateTime.now(), unreadMessages: 0),
+        );
+
+        await tester.drag(find.byType(StreamMessageListView), const Offset(0, 400));
+        await tester.pumpAndSettle();
+
+        final fromOther = Message(
+          id: 'new-from-other-reset-probe',
+          text: 'Out of view',
+          user: other,
+          createdAt: DateTime.now(),
+        );
+        await deliverMessageNew(tester, newMessage: fromOther, existing: messages);
+
+        expect(badgeLabel(tester), '1');
+
+        // Scroll back down to the bottom.
+        await tester.drag(find.byType(StreamMessageListView), const Offset(0, -1000));
+        await tester.pumpAndSettle();
+
+        // The scroll-to-bottom button itself hides at the bottom, so the
+        // badge is gone too.
+        expect(badgeLabel(tester), isNull);
+
+        // Scrolling away from the bottom again, with no further arrivals in
+        // between, must show the button with no badge — the earlier count
+        // should have been cleared on reaching the bottom, not just hidden.
+        await tester.drag(find.byType(StreamMessageListView), const Offset(0, 400));
+        await tester.pumpAndSettle();
+
+        expect(badgeLabel(tester), isNull);
+      },
+    );
+  });
+}
diff --git a/packages/stream_chat_flutter/test/src/mocks.dart b/packages/stream_chat_flutter/test/src/mocks.dart
index 9b1ac43b9a..cab7f0036a 100644
--- a/packages/stream_chat_flutter/test/src/mocks.dart
+++ b/packages/stream_chat_flutter/test/src/mocks.dart
@@ -90,6 +90,7 @@ class MockChannelState extends Mock implements ChannelClientState {
     when(() => typingEventsStream).thenAnswer((_) => Stream.value({}));
     when(() => unreadCount).thenReturn(0);
     when(() => isUpToDate).thenReturn(true);
+    when(() => isMarkedAsUnread).thenReturn(false);
     when(() => read).thenReturn([]);
     when(() => draftStream).thenAnswer((_) => Stream.value(null));
     when(() => threadDraftStream(any())).thenAnswer((_) => Stream.value(null));

From 452ed9be28de4f7ae96a2ae7a7908b13d5a08c62 Mon Sep 17 00:00:00 2001
From: Rene Floor <rene.floor@getstream.io>
Date: Thu, 6 Aug 2026 12:45:35 +0200
Subject: [PATCH 4/7] feat(i18n): add unreadMessagesSeparatorLabel translations

Adds the new count-aware label (Translations.unreadMessagesSeparatorLabel,
introduced in stream_chat_flutter) across all 11 supported locales,
plus the add_new_lang.dart example template and test coverage.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
---
 packages/stream_chat_localizations/CHANGELOG.md             | 1 +
 .../stream_chat_localizations/example/lib/add_new_lang.dart | 6 ++++++
 .../lib/src/stream_chat_localizations_ca.dart               | 6 ++++++
 .../lib/src/stream_chat_localizations_de.dart               | 6 ++++++
 .../lib/src/stream_chat_localizations_en.dart               | 6 ++++++
 .../lib/src/stream_chat_localizations_es.dart               | 6 ++++++
 .../lib/src/stream_chat_localizations_fr.dart               | 6 ++++++
 .../lib/src/stream_chat_localizations_hi.dart               | 6 ++++++
 .../lib/src/stream_chat_localizations_it.dart               | 6 ++++++
 .../lib/src/stream_chat_localizations_ja.dart               | 5 +++++
 .../lib/src/stream_chat_localizations_ko.dart               | 5 +++++
 .../lib/src/stream_chat_localizations_no.dart               | 6 ++++++
 .../lib/src/stream_chat_localizations_pt.dart               | 6 ++++++
 .../stream_chat_localizations/test/translations_test.dart   | 3 ++-
 14 files changed, 73 insertions(+), 1 deletion(-)

diff --git a/packages/stream_chat_localizations/CHANGELOG.md b/packages/stream_chat_localizations/CHANGELOG.md
index d583d86b58..2c6bc6bc56 100644
--- a/packages/stream_chat_localizations/CHANGELOG.md
+++ b/packages/stream_chat_localizations/CHANGELOG.md
@@ -3,6 +3,7 @@
 ✅ Added
 
 - Added connection-error translations (`connectionErrorTitle`/`Description`, `slowConnectionErrorTitle`/`Description`, `genericErrorTitle`/`Description`) for all supported locales.
+- Added `unreadMessagesSeparatorLabel` for all supported locales.
 
 ## 10.2.0
 
diff --git a/packages/stream_chat_localizations/example/lib/add_new_lang.dart b/packages/stream_chat_localizations/example/lib/add_new_lang.dart
index 0e3fcfe28a..bf0edb489b 100644
--- a/packages/stream_chat_localizations/example/lib/add_new_lang.dart
+++ b/packages/stream_chat_localizations/example/lib/add_new_lang.dart
@@ -510,6 +510,12 @@ class NnStreamChatLocalizations extends GlobalStreamChatLocalizations {
   @override
   String unreadMessagesSeparatorText() => 'New messages';
 
+  @override
+  String unreadMessagesSeparatorLabel({required int count}) {
+    if (count == 1) return '1 unread message';
+    return '$count unread messages';
+  }
+
   @override
   String get enableFileAccessMessage => 'Enable file access to continue';
 
diff --git a/packages/stream_chat_localizations/lib/src/stream_chat_localizations_ca.dart b/packages/stream_chat_localizations/lib/src/stream_chat_localizations_ca.dart
index 063cd39e74..abedc50ec2 100644
--- a/packages/stream_chat_localizations/lib/src/stream_chat_localizations_ca.dart
+++ b/packages/stream_chat_localizations/lib/src/stream_chat_localizations_ca.dart
@@ -495,6 +495,12 @@ class StreamChatLocalizationsCa extends GlobalStreamChatLocalizations {
   @override
   String unreadMessagesSeparatorText() => 'Missatges nous';
 
+  @override
+  String unreadMessagesSeparatorLabel({required int count}) {
+    if (count == 1) return '1 missatge no llegit';
+    return '$count missatges no llegits';
+  }
+
   @override
   String get enableFileAccessMessage => "Habilita l'accés als fitxers per poder compartir-los amb amics";
 
diff --git a/packages/stream_chat_localizations/lib/src/stream_chat_localizations_de.dart b/packages/stream_chat_localizations/lib/src/stream_chat_localizations_de.dart
index d0be1bc668..dac3b8249b 100644
--- a/packages/stream_chat_localizations/lib/src/stream_chat_localizations_de.dart
+++ b/packages/stream_chat_localizations/lib/src/stream_chat_localizations_de.dart
@@ -492,6 +492,12 @@ class StreamChatLocalizationsDe extends GlobalStreamChatLocalizations {
   @override
   String unreadMessagesSeparatorText() => 'Neue Nachrichten';
 
+  @override
+  String unreadMessagesSeparatorLabel({required int count}) {
+    if (count == 1) return '1 ungelesene Nachricht';
+    return '$count ungelesene Nachrichten';
+  }
+
   @override
   String get enableFileAccessMessage =>
       'Bitte aktivieren Sie den Zugriff auf Dateien, damit Sie sie mit Freunden teilen können.';
diff --git a/packages/stream_chat_localizations/lib/src/stream_chat_localizations_en.dart b/packages/stream_chat_localizations/lib/src/stream_chat_localizations_en.dart
index 09caf5330a..9c925ea9f3 100644
--- a/packages/stream_chat_localizations/lib/src/stream_chat_localizations_en.dart
+++ b/packages/stream_chat_localizations/lib/src/stream_chat_localizations_en.dart
@@ -494,6 +494,12 @@ class StreamChatLocalizationsEn extends GlobalStreamChatLocalizations {
   @override
   String unreadMessagesSeparatorText() => 'New messages';
 
+  @override
+  String unreadMessagesSeparatorLabel({required int count}) {
+    if (count == 1) return '1 unread message';
+    return '$count unread messages';
+  }
+
   @override
   String get enableFileAccessMessage => 'Please enable access to files so you can share them with friends.';
 
diff --git a/packages/stream_chat_localizations/lib/src/stream_chat_localizations_es.dart b/packages/stream_chat_localizations/lib/src/stream_chat_localizations_es.dart
index 2f90d3f4ed..bac21258e0 100644
--- a/packages/stream_chat_localizations/lib/src/stream_chat_localizations_es.dart
+++ b/packages/stream_chat_localizations/lib/src/stream_chat_localizations_es.dart
@@ -498,6 +498,12 @@ No es posible añadir más de $limit archivos adjuntos
   @override
   String unreadMessagesSeparatorText() => 'Nuevos mensajes';
 
+  @override
+  String unreadMessagesSeparatorLabel({required int count}) {
+    if (count == 1) return '1 mensaje no leído';
+    return '$count mensajes no leídos';
+  }
+
   @override
   String get enableFileAccessMessage => 'Habilite el acceso a los archivos para poder compartirlos con amigos.';
 
diff --git a/packages/stream_chat_localizations/lib/src/stream_chat_localizations_fr.dart b/packages/stream_chat_localizations/lib/src/stream_chat_localizations_fr.dart
index 77d17595f4..35f2e1af68 100644
--- a/packages/stream_chat_localizations/lib/src/stream_chat_localizations_fr.dart
+++ b/packages/stream_chat_localizations/lib/src/stream_chat_localizations_fr.dart
@@ -498,6 +498,12 @@ Limite de pièces jointes dépassée : il n'est pas possible d'ajouter plus de $
   @override
   String unreadMessagesSeparatorText() => 'Nouveaux messages';
 
+  @override
+  String unreadMessagesSeparatorLabel({required int count}) {
+    if (count == 1) return '1 message non lu';
+    return '$count messages non lus';
+  }
+
   @override
   String get enableFileAccessMessage =>
       "Veuillez autoriser l'accès aux fichiers afin de pouvoir les partager avec des amis.";
diff --git a/packages/stream_chat_localizations/lib/src/stream_chat_localizations_hi.dart b/packages/stream_chat_localizations/lib/src/stream_chat_localizations_hi.dart
index eaaea55d36..d6ae8a2fdb 100644
--- a/packages/stream_chat_localizations/lib/src/stream_chat_localizations_hi.dart
+++ b/packages/stream_chat_localizations/lib/src/stream_chat_localizations_hi.dart
@@ -496,6 +496,12 @@ class StreamChatLocalizationsHi extends GlobalStreamChatLocalizations {
   @override
   String unreadMessagesSeparatorText() => 'नए संदेश।';
 
+  @override
+  String unreadMessagesSeparatorLabel({required int count}) {
+    if (count == 1) return '1 अपठित संदेश';
+    return '$count अपठित संदेश';
+  }
+
   @override
   String get enableFileAccessMessage => 'कृपया फ़ाइलों तक पहुंच सक्षम करें ताकि आप उन्हें मित्रों के साथ साझा कर सकें।';
 
diff --git a/packages/stream_chat_localizations/lib/src/stream_chat_localizations_it.dart b/packages/stream_chat_localizations/lib/src/stream_chat_localizations_it.dart
index 1510cbab32..6986d8c5e4 100644
--- a/packages/stream_chat_localizations/lib/src/stream_chat_localizations_it.dart
+++ b/packages/stream_chat_localizations/lib/src/stream_chat_localizations_it.dart
@@ -501,6 +501,12 @@ Attenzione: il limite massimo di $limit file è stato superato.
   @override
   String unreadMessagesSeparatorText() => 'Nuovi messaggi';
 
+  @override
+  String unreadMessagesSeparatorLabel({required int count}) {
+    if (count == 1) return '1 messaggio non letto';
+    return '$count messaggi non letti';
+  }
+
   @override
   String get enableFileAccessMessage =>
       "Per favore attiva l'accesso ai file cosí potrai condividerli con i tuoi amici.";
diff --git a/packages/stream_chat_localizations/lib/src/stream_chat_localizations_ja.dart b/packages/stream_chat_localizations/lib/src/stream_chat_localizations_ja.dart
index 3e443c6f05..0dafeeefb0 100644
--- a/packages/stream_chat_localizations/lib/src/stream_chat_localizations_ja.dart
+++ b/packages/stream_chat_localizations/lib/src/stream_chat_localizations_ja.dart
@@ -484,6 +484,11 @@ class StreamChatLocalizationsJa extends GlobalStreamChatLocalizations {
   @override
   String unreadMessagesSeparatorText() => '新しいメッセージ。';
 
+  @override
+  String unreadMessagesSeparatorLabel({required int count}) {
+    return '未読メッセージ $count 件';
+  }
+
   @override
   String get enableFileAccessMessage => '友達と共有できるように、ファイルへのアクセスを有効にしてください。';
 
diff --git a/packages/stream_chat_localizations/lib/src/stream_chat_localizations_ko.dart b/packages/stream_chat_localizations/lib/src/stream_chat_localizations_ko.dart
index f22a77c352..4c06b62894 100644
--- a/packages/stream_chat_localizations/lib/src/stream_chat_localizations_ko.dart
+++ b/packages/stream_chat_localizations/lib/src/stream_chat_localizations_ko.dart
@@ -485,6 +485,11 @@ class StreamChatLocalizationsKo extends GlobalStreamChatLocalizations {
   @override
   String unreadMessagesSeparatorText() => '새 메시지.';
 
+  @override
+  String unreadMessagesSeparatorLabel({required int count}) {
+    return '읽지 않은 메시지 $count개';
+  }
+
   @override
   String get enableFileAccessMessage => '친구와 공유할 수 있도록 파일에 대한 액세스를 허용하세요.';
 
diff --git a/packages/stream_chat_localizations/lib/src/stream_chat_localizations_no.dart b/packages/stream_chat_localizations/lib/src/stream_chat_localizations_no.dart
index 180160c55d..6141413485 100644
--- a/packages/stream_chat_localizations/lib/src/stream_chat_localizations_no.dart
+++ b/packages/stream_chat_localizations/lib/src/stream_chat_localizations_no.dart
@@ -436,6 +436,12 @@ class StreamChatLocalizationsNo extends GlobalStreamChatLocalizations {
   @override
   String unreadMessagesSeparatorText() => 'Nye meldinger.';
 
+  @override
+  String unreadMessagesSeparatorLabel({required int count}) {
+    if (count == 1) return '1 ulest melding';
+    return '$count uleste meldinger';
+  }
+
   @override
   String get couldNotReadBytesFromFileError => 'Kunne ikke lese bytes fra filen.';
 
diff --git a/packages/stream_chat_localizations/lib/src/stream_chat_localizations_pt.dart b/packages/stream_chat_localizations/lib/src/stream_chat_localizations_pt.dart
index 5909875c1f..23e2c78e4f 100644
--- a/packages/stream_chat_localizations/lib/src/stream_chat_localizations_pt.dart
+++ b/packages/stream_chat_localizations/lib/src/stream_chat_localizations_pt.dart
@@ -497,6 +497,12 @@ Não é possível adicionar mais de $limit arquivos de uma vez
   @override
   String unreadMessagesSeparatorText() => 'Novas mensagens';
 
+  @override
+  String unreadMessagesSeparatorLabel({required int count}) {
+    if (count == 1) return '1 mensagem não lida';
+    return '$count mensagens não lidas';
+  }
+
   @override
   String get enableFileAccessMessage => 'Ative o acesso aos arquivos para poder compartilhá-los com amigos.';
 
diff --git a/packages/stream_chat_localizations/test/translations_test.dart b/packages/stream_chat_localizations/test/translations_test.dart
index a763436c90..9200af68e6 100644
--- a/packages/stream_chat_localizations/test/translations_test.dart
+++ b/packages/stream_chat_localizations/test/translations_test.dart
@@ -229,7 +229,8 @@ void main() {
       expect(localizations.enableFileAccessMessage, isNotNull);
       expect(localizations.allowFileAccessMessage, isNotNull);
       expect(localizations.unreadCountIndicatorLabel(unreadCount: 2), isNotNull);
-      expect(localizations.unreadMessagesSeparatorText(), isNotNull);
+      expect(localizations.unreadMessagesSeparatorLabel(count: 1), isNotNull);
+      expect(localizations.unreadMessagesSeparatorLabel(count: 2), isNotNull);
       expect(localizations.markUnreadError, isNotNull);
       expect(localizations.markAsUnreadLabel, isNotNull);
       // Create poll

From 5c3e35ec5b1d21fa5834256561c70b6198c49af6 Mon Sep 17 00:00:00 2001
From: Rene Floor <rene.floor@getstream.io>
Date: Thu, 6 Aug 2026 16:09:02 +0200
Subject: [PATCH 5/7] test improvements

---
 .../test/src/client/channel_test.dart         | 142 ++++++++++++++++++
 1 file changed, 142 insertions(+)

diff --git a/packages/stream_chat/test/src/client/channel_test.dart b/packages/stream_chat/test/src/client/channel_test.dart
index d3330511d8..d550c8fcbc 100644
--- a/packages/stream_chat/test/src/client/channel_test.dart
+++ b/packages/stream_chat/test/src/client/channel_test.dart
@@ -6786,6 +6786,120 @@ void main() {
         },
       );
 
+      group('isMarkedAsUnread', () {
+        setUp(() {
+          // A message.read event from the current user also reconciles
+          // delivery status — stub it so that call doesn't throw.
+          when(
+            () => client.channelDeliveryReporter.reconcileDelivery(any()),
+          ).thenAnswer((_) async {});
+        });
+
+        test('defaults to false', () {
+          expect(channel.state?.isMarkedAsUnread, isFalse);
+        });
+
+        test(
+          'is set by a notification.mark_unread event from the current user',
+          () async {
+            final currentUser = client.state.currentUser!;
+
+            final markUnreadEvent = Event(
+              cid: channel.cid,
+              type: EventType.notificationMarkUnread,
+              user: currentUser,
+              lastReadAt: DateTime(2019),
+              unreadMessages: 5,
+            );
+            client.addEvent(markUnreadEvent);
+            await Future.delayed(Duration.zero);
+
+            expect(channel.state?.isMarkedAsUnread, isTrue);
+          },
+        );
+
+        test(
+          'is NOT set by a notification.mark_unread event from a different user',
+          () async {
+            final markUnreadEvent = Event(
+              cid: channel.cid,
+              type: EventType.notificationMarkUnread,
+              user: User(id: 'someone-else'),
+              lastReadAt: DateTime(2019),
+              unreadMessages: 5,
+            );
+            client.addEvent(markUnreadEvent);
+            await Future.delayed(Duration.zero);
+
+            expect(channel.state?.isMarkedAsUnread, isFalse);
+          },
+        );
+
+        test(
+          'is cleared by a message.read event from the current user',
+          () async {
+            final currentUser = client.state.currentUser!;
+
+            client.addEvent(
+              Event(
+                cid: channel.cid,
+                type: EventType.notificationMarkUnread,
+                user: currentUser,
+                lastReadAt: DateTime(2019),
+                unreadMessages: 5,
+              ),
+            );
+            await Future.delayed(Duration.zero);
+            expect(channel.state?.isMarkedAsUnread, isTrue);
+
+            client.addEvent(
+              Event(
+                cid: channel.cid,
+                type: EventType.messageRead,
+                user: currentUser,
+                createdAt: DateTime(2022),
+                unreadMessages: 0,
+              ),
+            );
+            await Future.delayed(Duration.zero);
+
+            expect(channel.state?.isMarkedAsUnread, isFalse);
+          },
+        );
+
+        test(
+          'is NOT cleared by a message.read event from a different user',
+          () async {
+            final currentUser = client.state.currentUser!;
+
+            client.addEvent(
+              Event(
+                cid: channel.cid,
+                type: EventType.notificationMarkUnread,
+                user: currentUser,
+                lastReadAt: DateTime(2019),
+                unreadMessages: 5,
+              ),
+            );
+            await Future.delayed(Duration.zero);
+            expect(channel.state?.isMarkedAsUnread, isTrue);
+
+            client.addEvent(
+              Event(
+                cid: channel.cid,
+                type: EventType.messageRead,
+                user: User(id: 'someone-else'),
+                createdAt: DateTime(2022),
+                unreadMessages: 0,
+              ),
+            );
+            await Future.delayed(Duration.zero);
+
+            expect(channel.state?.isMarkedAsUnread, isTrue);
+          },
+        );
+      });
+
       test('should update read state on message delivered event', () async {
         final currentUser = User(id: 'test-user');
         final distantPast = DateTime.fromMillisecondsSinceEpoch(0, isUtc: true);
@@ -10775,6 +10889,34 @@ void main() {
       },
     );
 
+    test(
+      'markUnreadByTimestamp sets isMarkedAsUnread locally',
+      () async {
+        final channel = _createLivestreamChannel();
+        expect(channel.state?.isMarkedAsUnread, isFalse);
+
+        await expectLater(
+          channel.markUnreadByTimestamp(DateTime(2024, 1, 1)),
+          completes,
+        );
+
+        expect(channel.state?.isMarkedAsUnread, isTrue);
+      },
+    );
+
+    test(
+      'markRead clears isMarkedAsUnread locally',
+      () async {
+        final channel = _createLivestreamChannel();
+        await channel.markUnreadByTimestamp(DateTime(2024, 1, 1));
+        expect(channel.state?.isMarkedAsUnread, isTrue);
+
+        await expectLater(channel.markRead(), completes);
+
+        expect(channel.state?.isMarkedAsUnread, isFalse);
+      },
+    );
+
     group('local read boundary anchors', () {
       final start = DateTime(2024, 1, 1);
       final messages = [

From 0808f07d651022e9d8f5f777a2cc50987a78b4fd Mon Sep 17 00:00:00 2001
From: Rene Floor <rene.floor@getstream.io>
Date: Fri, 7 Aug 2026 10:03:32 +0200
Subject: [PATCH 6/7] fix review comments

---
 .../message_list_view/mark_read_details.dart  |  2 +
 .../message_list_view/message_list_view.dart  | 47 +++++++++++-----
 ...tream_message_list_view_configuration.dart |  6 +++
 .../src/message_list_view/mark_read_test.dart | 54 ++++++++++++++++++-
 .../lib/src/stream_channel.dart               |  3 ++
 5 files changed, 97 insertions(+), 15 deletions(-)

diff --git a/packages/stream_chat_flutter/lib/src/message_list_view/mark_read_details.dart b/packages/stream_chat_flutter/lib/src/message_list_view/mark_read_details.dart
index d2b4fda27d..b0d9a7e660 100644
--- a/packages/stream_chat_flutter/lib/src/message_list_view/mark_read_details.dart
+++ b/packages/stream_chat_flutter/lib/src/message_list_view/mark_read_details.dart
@@ -1,3 +1,5 @@
+import 'package:stream_chat_flutter/stream_chat_flutter.dart';
+
 /// The information available when deciding whether to automatically mark a
 /// [StreamMessageListView]'s channel as read.
 ///
diff --git a/packages/stream_chat_flutter/lib/src/message_list_view/message_list_view.dart b/packages/stream_chat_flutter/lib/src/message_list_view/message_list_view.dart
index e84be35481..07779e0ef4 100644
--- a/packages/stream_chat_flutter/lib/src/message_list_view/message_list_view.dart
+++ b/packages/stream_chat_flutter/lib/src/message_list_view/message_list_view.dart
@@ -357,7 +357,13 @@ class _StreamMessageListViewState extends State<StreamMessageListView> {
   // burned on capturing the baseline instead of acting on it. Cleared once
   // a mark-read actually goes through, or once `isMarkedAsUnread` itself
   // clears (so a future mark-unread starts its own fresh snapshot).
-  Iterable<ItemPosition>? _markUnreadViewportSnapshot;
+  //
+  // Holds visible item *indices* rather than full [ItemPosition]s: comparing
+  // full positions would latch divergence on a sub-pixel edge change from an
+  // unrelated relayout (async attachment sizing, keyboard inset, image
+  // load) even though the user never scrolled, undoing the manual
+  // mark-unread almost instantly.
+  List<int>? _markUnreadViewportSnapshot;
 
   // Sticky once true: sighted the first time [_handleItemPositionsChanged]
   // (or, as a fallback, [_maybeMarkMessagesAsRead] itself) sees item
@@ -424,7 +430,7 @@ class _StreamMessageListViewState extends State<StreamMessageListView> {
       // read-stream emission while still marked unread shouldn't keep
       // chasing the latest position and never let a genuine scroll differ
       // from it.
-      _markUnreadViewportSnapshot ??= _itemPositionListener.itemPositions.value.toList();
+      _markUnreadViewportSnapshot ??= _itemPositionListener.itemPositions.value.map((it) => it.index).toList();
     } else {
       _markUnreadViewportSnapshot = null;
       _markUnreadViewportDiverged = false;
@@ -603,7 +609,10 @@ class _StreamMessageListViewState extends State<StreamMessageListView> {
     _highlightState.value = (id: messageId, generation: _highlightState.value.generation + 1);
   }
 
-  Future<void> _scrollToMessage({
+  // Returns whether the list actually scrolled to `messageId` — `false` for
+  // any of the bail-out paths below (target not found even after
+  // pagination, widget unmounted mid-pagination, or the SPL not attached).
+  Future<bool> _scrollToMessage({
     required String messageId,
     double alignment = 0.5, // center the message in the viewport by default
     bool highlight = true,
@@ -617,7 +626,7 @@ class _StreamMessageListViewState extends State<StreamMessageListView> {
     if (index < 0) {
       // No around-reply pagination in thread mode yet — bail rather than
       // clobber the parent channel's loaded window.
-      if (_isThreadConversation) return;
+      if (_isThreadConversation) return false;
 
       // Target isn't in the loaded channel window. Paginate around it, wait
       // one frame for the BetterStreamBuilder rebuild to flush `messages`,
@@ -625,17 +634,17 @@ class _StreamMessageListViewState extends State<StreamMessageListView> {
       // `_buildListView` on each emission, so an index captured before the
       // await would be stale.
       await streamChannel!.loadChannelAtMessage(messageId);
-      if (!mounted) return;
+      if (!mounted) return false;
       await WidgetsBinding.instance.endOfFrame;
-      if (!mounted) return;
+      if (!mounted) return false;
       index = messages.indexWhere((m) => m.id == messageId);
-      if (index < 0) return;
+      if (index < 0) return false;
     }
 
     // Bail when the SPL isn't attached — `scrollTo` would throw, and
     // highlighting an off-screen message is meaningless.
     final controller = _scrollController;
-    if (controller == null || !controller.isAttached) return;
+    if (controller == null || !controller.isAttached) return false;
 
     // Wait for the scroll to settle before flagging the message as
     // highlighted; otherwise the highlight tween fires while the list is
@@ -647,6 +656,7 @@ class _StreamMessageListViewState extends State<StreamMessageListView> {
     );
 
     if (highlight && mounted) _highlightMessage(messageId);
+    return true;
   }
 
   // Wraps [child] in the highlight pulse if [message] is the currently
@@ -1105,12 +1115,16 @@ class _StreamMessageListViewState extends State<StreamMessageListView> {
     final anchorId = _unreadDivider.value.anchorId ?? _unreadBaseline?.lastReadMessageId;
     if (anchorId == null) return;
 
-    _hasSeenFirstUnread.value = true;
     // Delegates to [_scrollToMessage], which falls back to
     // [StreamChannelState.loadChannelAtMessage] when the anchor isn't in the
     // currently loaded window — after which the real anchor resolves
     // naturally via the retry in [_buildListView], rendering divider A too.
-    await _scrollToMessage(messageId: anchorId, highlight: false);
+    final didJump = await _scrollToMessage(messageId: anchorId, highlight: false);
+    // Only claim the boundary as seen once the jump actually landed —
+    // otherwise (message not found even after pagination, or the SPL not
+    // attached) the pill would vanish and the mark-read gate would open for
+    // a boundary the user never actually reached.
+    if (didJump && mounted) _hasSeenFirstUnread.value = true;
   }
 
   Future<void> _onUnreadPillDismissTap() async {
@@ -1440,15 +1454,22 @@ class _StreamMessageListViewState extends State<StreamMessageListView> {
   // unchanged and re-block a mark-read the round trip should already have
   // earned. Safe to call on every position-changed tick — a no-op once
   // already diverged.
+  //
+  // Compares the set of visible item *indices* rather than full
+  // [ItemPosition]s (which also carry leading/trailing edge offsets) — an
+  // unrelated relayout that nudges an edge by a fraction of a pixel isn't
+  // evidence the user did anything, and shouldn't count as divergence.
   void _checkMarkUnreadViewportDivergence(Iterable<ItemPosition> itemPositions) {
+    final visibleIndices = itemPositions.map((it) => it.index).toList();
+
     if (_markUnreadViewportSnapshot == null) {
-      _markUnreadViewportSnapshot = itemPositions.toList();
+      _markUnreadViewportSnapshot = visibleIndices;
       return;
     }
     if (_markUnreadViewportDiverged) return;
 
-    const positionsEquality = UnorderedIterableEquality<ItemPosition>();
-    if (!positionsEquality.equals(itemPositions, _markUnreadViewportSnapshot)) {
+    const indicesEquality = UnorderedIterableEquality<int>();
+    if (!indicesEquality.equals(visibleIndices, _markUnreadViewportSnapshot)) {
       _markUnreadViewportDiverged = true;
     }
   }
diff --git a/packages/stream_chat_flutter/lib/src/message_list_view/stream_message_list_view_configuration.dart b/packages/stream_chat_flutter/lib/src/message_list_view/stream_message_list_view_configuration.dart
index 263fd59bf0..afdf1db723 100644
--- a/packages/stream_chat_flutter/lib/src/message_list_view/stream_message_list_view_configuration.dart
+++ b/packages/stream_chat_flutter/lib/src/message_list_view/stream_message_list_view_configuration.dart
@@ -55,6 +55,12 @@ class StreamMessageListViewConfiguration {
   ///
   /// Only affects channel reads; has no effect on thread reads or on
   /// [markReadWhenAtTheBottom] being `false`.
+  ///
+  /// Participates in this configuration's `==`/`hashCode`, so an inline
+  /// closure gives every rebuild a new identity and can make otherwise
+  /// identical configurations compare unequal. Hosts that rely on
+  /// configuration equality should hoist the predicate into a field or a
+  /// static function instead.
   final StreamShouldMarkReadPredicate? shouldMarkRead;
 
   /// Whether swiping a message triggers a quoted-reply action.
diff --git a/packages/stream_chat_flutter/test/src/message_list_view/mark_read_test.dart b/packages/stream_chat_flutter/test/src/message_list_view/mark_read_test.dart
index 27efe5dda3..fa3fc95e9d 100644
--- a/packages/stream_chat_flutter/test/src/message_list_view/mark_read_test.dart
+++ b/packages/stream_chat_flutter/test/src/message_list_view/mark_read_test.dart
@@ -330,16 +330,29 @@ void main() {
       (tester) async {
         final other = User(id: 'otherid');
         final messages = generateConversation(20, users: [other]).reversed.toList();
+        StreamMarkReadDetails? capturedDetails;
 
         await pumpMessageList(
           tester,
           messages: messages,
           isUpToDate: true,
           unreadCount: 5,
-          shouldMarkRead: (details) => false,
+          shouldMarkRead: (details) {
+            capturedDetails = details;
+            return false;
+          },
         );
 
         verifyNever(() => channel.markRead(messageId: any(named: 'messageId')));
+
+        // Opened at the bottom with nothing pre-existing unread and no
+        // active manual mark-unread — the default gating would have
+        // allowed this; only the override blocks it.
+        expect(capturedDetails, isNotNull);
+        expect(capturedDetails!.unreadCount, 5);
+        expect(capturedDetails!.hasSeenLastMessage, isTrue);
+        expect(capturedDetails!.hasSeenFirstUnreadMessage, isTrue);
+        expect(capturedDetails!.isMarkedAsUnread, isFalse);
       },
     );
 
@@ -349,6 +362,7 @@ void main() {
         final other = User(id: 'otherid');
         final messages = generateConversation(20, users: [other]).reversed.toList();
         final lastReadMessageId = messages[10].id;
+        StreamMarkReadDetails? capturedDetails;
 
         await pumpMessageList(
           tester,
@@ -362,10 +376,46 @@ void main() {
             unreadMessages: 5,
             lastReadMessageId: lastReadMessageId,
           ),
-          shouldMarkRead: (details) => true,
+          shouldMarkRead: (details) {
+            capturedDetails = details;
+            return true;
+          },
         );
 
         verify(() => channel.markRead(messageId: any(named: 'messageId'))).called(1);
+
+        // The unseen pre-existing unread boundary is exactly what the
+        // default gating would have blocked on; the override allows it
+        // anyway.
+        expect(capturedDetails, isNotNull);
+        expect(capturedDetails!.unreadCount, 5);
+        expect(capturedDetails!.hasSeenFirstUnreadMessage, isFalse);
+        expect(capturedDetails!.isMarkedAsUnread, isFalse);
+      },
+    );
+
+    testWidgets(
+      'a shouldMarkRead override sees isMarkedAsUnread when the channel has an active manual mark-unread',
+      (tester) async {
+        final other = User(id: 'otherid');
+        final messages = generateConversation(20, users: [other]).reversed.toList();
+        when(() => channelClientState.isMarkedAsUnread).thenReturn(true);
+        StreamMarkReadDetails? capturedDetails;
+
+        await pumpMessageList(
+          tester,
+          messages: messages,
+          isUpToDate: true,
+          unreadCount: 5,
+          shouldMarkRead: (details) {
+            capturedDetails = details;
+            return false;
+          },
+        );
+
+        expect(capturedDetails, isNotNull);
+        expect(capturedDetails!.isMarkedAsUnread, isTrue);
+        expect(capturedDetails!.unreadCount, 5);
       },
     );
   });
diff --git a/packages/stream_chat_flutter_core/lib/src/stream_channel.dart b/packages/stream_chat_flutter_core/lib/src/stream_channel.dart
index 5e4232f356..eb0b6a7dc7 100644
--- a/packages/stream_chat_flutter_core/lib/src/stream_channel.dart
+++ b/packages/stream_chat_flutter_core/lib/src/stream_channel.dart
@@ -88,6 +88,9 @@ class StreamChannel extends StatefulWidget {
   ///
   /// Has no effect on [StreamChannel.value], which never repositions the
   /// loaded window.
+  ///
+  /// Only read once, during channel initialization — changing it after this
+  /// widget has mounted does not reposition the current viewport.
   final bool openAtFirstUnread;
 
   /// Widget builder used while the channel is initialising.

From 27e1a51a76b3208cd88d08d052e0ca278e6e1161 Mon Sep 17 00:00:00 2001
From: Rene Floor <rene.floor@getstream.io>
Date: Fri, 7 Aug 2026 10:06:07 +0200
Subject: [PATCH 7/7] fix(ui): dispose _showScrollToBottom notifier

Missed in the previous review-comment pass; it's created alongside the
other mark-read/unread notifiers and needs the same teardown.
---
 .../lib/src/message_list_view/message_list_view.dart             | 1 +
 1 file changed, 1 insertion(+)

diff --git a/packages/stream_chat_flutter/lib/src/message_list_view/message_list_view.dart b/packages/stream_chat_flutter/lib/src/message_list_view/message_list_view.dart
index 07779e0ef4..78e590f276 100644
--- a/packages/stream_chat_flutter/lib/src/message_list_view/message_list_view.dart
+++ b/packages/stream_chat_flutter/lib/src/message_list_view/message_list_view.dart
@@ -593,6 +593,7 @@ class _StreamMessageListViewState extends State<StreamMessageListView> {
     _unreadDividerGrowth.dispose();
     _hasSeenFirstUnread.dispose();
     _scrollToBottomBadge.dispose();
+    _showScrollToBottom.dispose();
     _highlightState.dispose();
     super.dispose();
   }