fix(mobile): mitigate message-post delay with optimistic rendering (#3037)

### What changed?

Adds optimistic rendering for newly locally posted messages in channels
and threads.
- Messages appear immediately
- Relay echoes and history are deduplicated by event ID
- Rejected or timed-out publishes erase the optimistic rendering.

The implementation covers reconnect and hydration races, channel-window
and legacy WebSocket paths, thread-local overlays, and rapid concurrent
sends.

### Why?

This is a valuable partial mitigation for
[BOT-1449](https://linear.app/squareup/issue/BOT-1449/buzz-mobile-posted-messages-dont-appear-until-leavingre-entering-the):
senders no longer depend on receiving a relay echo before seeing their
own post.

It does not address the likely primary cause of stale channels. Mobile
currently does not recover live subscriptions after a rate-limited relay
`CLOSED`; that recovery is being handled separately.

### How is it tested?

Full mobile suite: 676 passed, 1 skipped.

Added regression coverage for optimistic insertion, authoritative
deduplication, rollback, reconnect and hydration, thread replies, rapid
and equal-time sends, never-echoed successful sends, and legacy
WebSocket retirement.

---------

Signed-off-by: npub102wg7q285p64ch2fjvstmf2ntn2sz3c4u5hmwatalc76mhsuauysftjtfj <7a9c8f0147a0755c5d499320bda5535cd5014715e52fb7757dfe3dadde1cef09@buzz.block.builderlab.xyz>
Co-authored-by: npub102wg7q285p64ch2fjvstmf2ntn2sz3c4u5hmwatalc76mhsuauysftjtfj <7a9c8f0147a0755c5d499320bda5535cd5014715e52fb7757dfe3dadde1cef09@buzz.block.builderlab.xyz>

mobile/lib/features/channels/channel_messages_provider.dart

@@ -3,6 +3,7 @@ import 'package:hooks_riverpod/hooks_riverpod.dart';
 
 import '../../shared/relay/relay.dart';
 import 'channel_management_provider.dart';
+import 'pending_local_messages_provider.dart';
 import 'channel_window.dart';
 import 'thread_replies_provider.dart';
 
@@ -87,6 +88,7 @@ class ChannelMessagesNotifier extends Notifier<AsyncValue<List<NostrEvent>>> {
 
       final history = await _fetchNewestHistory(session);
       if (!_isCurrentInit(initVersion)) return;
+      _confirmLocalMessages(history.map((event) => event.id));
 
       final existing = state.value ?? const <NostrEvent>[];
       final existingIds = existing.map((event) => event.id).toSet();
@@ -159,7 +161,13 @@ class ChannelMessagesNotifier extends Notifier<AsyncValue<List<NostrEvent>>> {
     },
   );
 
-  void _handleLiveEvent(NostrEvent event) {
+  void _handleLiveEvent(NostrEvent event, {bool authoritative = true}) {
+    // Reply ownership and its thread-local overlay must transition together.
+    // The authoritative thread query performs both confirmations after it
+    // contains the reply; a live echo only triggers that query below.
+    if (authoritative && event.threadReference.parentId == null) {
+      _confirmLocalMessages([event.id]);
+    }
     if (_usingChannelWindow) {
       _handleWindowLiveEvent(event);
     } else {
@@ -223,19 +231,108 @@ class ChannelMessagesNotifier extends Notifier<AsyncValue<List<NostrEvent>>> {
     return true;
   }
 
+  void _confirmLocalMessages(Iterable<String> eventIds) {
+    ref
+        .read(pendingLocalMessagesProvider(channelId).notifier)
+        .confirm(eventIds);
+  }
+
   static bool _isMembershipEvent(String content) {
     return content.contains('member_joined') ||
         content.contains('member_left') ||
         content.contains('member_removed');
   }
 
+  /// Adds a just-signed outgoing message before the relay acknowledges it.
+  /// The live relay echo is deduplicated by event id.
+  void addLocalMessage(NostrEvent event) {
+    ref.read(pendingLocalMessagesProvider(channelId).notifier).add(event);
+    final thread = event.threadReference;
+    if (thread.parentId != null) {
+      final rootId = thread.rootId;
+      if (rootId == null) {
+        throw StateError('Reply ${event.id} has a parent but no thread root.');
+      }
+      ref
+          .read(
+            threadLocalRepliesProvider(
+              ThreadRepliesArgs(channelId: channelId, rootId: rootId),
+            ).notifier,
+          )
+          .add(event);
+      return;
+    }
+
+    final isTimelineRow = EventKind.channelTimelineContentKinds.contains(
+      event.kind,
+    );
+    if (!_usingChannelWindow && isTimelineRow) {
+      _windowStore = mergeLiveChannelWindowEvent(
+        _windowStore,
+        event,
+        isTimelineRow: true,
+      );
+    }
+    _handleLiveEvent(event, authoritative: false);
+  }
+
+  /// Releases rollback ownership after the publish future succeeds. The
+  /// optimistic row (and any thread overlay) remains visible until relay data
+  /// replaces it, because OK and EVENT delivery are unordered.
+  void completeLocalMessage(String eventId) {
+    _confirmLocalMessages([eventId]);
+  }
+
+  /// Rolls back a local message when its publish is rejected or times out.
+  void removeLocalMessage(String eventId) {
+    final pending = ref
+        .read(pendingLocalMessagesProvider(channelId).notifier)
+        .take(eventId);
+    if (pending == null) return;
+
+    final thread = pending.threadReference;
+    if (thread.parentId != null) {
+      final rootId = thread.rootId;
+      if (rootId == null) {
+        throw StateError('Reply $eventId has a parent but no thread root.');
+      }
+      ref
+          .read(
+            threadLocalRepliesProvider(
+              ThreadRepliesArgs(channelId: channelId, rootId: rootId),
+            ).notifier,
+          )
+          .remove(eventId);
+      return;
+    }
+
+    final nextOverlay = _windowStore.liveOverlay
+        .where((event) => event.id != eventId)
+        .toList();
+    if (nextOverlay.length != _windowStore.liveOverlay.length) {
+      _windowStore = ChannelWindowStore(
+        pages: _windowStore.pages,
+        liveOverlay: nextOverlay,
+        liveAux: _windowStore.liveAux,
+      );
+    }
+
+    final current = state.value ?? _lastKnownMessages ?? const <NostrEvent>[];
+    final next = current.where((event) => event.id != eventId).toList();
+    _lastKnownMessages = next;
+    state = AsyncData(next);
+  }
+
   static List<NostrEvent> _mergeEvent(
     List<NostrEvent> current,
     NostrEvent incoming,
   ) {
     if (current.any((e) => e.id == incoming.id)) return current;
     final updated = [...current, incoming];
-    updated.sort((a, b) => a.createdAt.compareTo(b.createdAt));
+    updated.sort((a, b) {
+      final createdAt = a.createdAt.compareTo(b.createdAt);
+      return createdAt != 0 ? createdAt : a.id.compareTo(b.id);
+    });
     return updated;
   }
 

mobile/lib/features/channels/pending_local_messages_provider.dart

@@ -0,0 +1,42 @@
+import 'package:hooks_riverpod/hooks_riverpod.dart';
+
+import '../../shared/relay/relay.dart';
+
+/// Signed local messages whose publish has not yet been corroborated by an
+/// authoritative relay EVENT or query result.
+class PendingLocalMessagesNotifier extends Notifier<Map<String, NostrEvent>> {
+  final String channelId;
+
+  PendingLocalMessagesNotifier(this.channelId);
+
+  @override
+  Map<String, NostrEvent> build() => const {};
+
+  void add(NostrEvent event) {
+    state = {...state, event.id: event};
+  }
+
+  NostrEvent? take(String eventId) {
+    final event = state[eventId];
+    if (event == null) return null;
+    final next = {...state}..remove(eventId);
+    state = next;
+    return event;
+  }
+
+  void confirm(Iterable<String> eventIds) {
+    final confirmed = eventIds.toSet();
+    if (!state.keys.any(confirmed.contains)) return;
+    state = {
+      for (final entry in state.entries)
+        if (!confirmed.contains(entry.key)) entry.key: entry.value,
+    };
+  }
+}
+
+final pendingLocalMessagesProvider =
+    NotifierProvider.family<
+      PendingLocalMessagesNotifier,
+      Map<String, NostrEvent>,
+      String
+    >(PendingLocalMessagesNotifier.new);

mobile/lib/features/channels/send_message_provider.dart

@@ -4,6 +4,7 @@ import '../../shared/relay/relay.dart';
 import '../channels/channel_management_provider.dart';
 import '../profile/user_cache_provider.dart';
 import '../profile/user_profile.dart';
+import 'channel_messages_provider.dart';
 
 /// Sends messages by signing an event with the user's nsec and publishing it
 /// over the relay's NIP-42-authenticated WebSocket session.
@@ -11,15 +12,25 @@ class SendMessage {
   final SignedEventRelay _signedEventRelay;
   final Future<List<ChannelMember>> Function(String channelId) _fetchMembers;
   final Map<String, UserProfile> Function() _readUserCache;
+  final void Function(String channelId, NostrEvent event) _addLocalMessage;
+  final void Function(String channelId, String eventId) _completeLocalMessage;
+  final void Function(String channelId, String eventId) _removeLocalMessage;
 
   SendMessage({
     required SignedEventRelay signedEventRelay,
     required Future<List<ChannelMember>> Function(String channelId)
     fetchMembers,
     required Map<String, UserProfile> Function() readUserCache,
+    required void Function(String channelId, NostrEvent event) addLocalMessage,
+    required void Function(String channelId, String eventId)
+    completeLocalMessage,
+    required void Function(String channelId, String eventId) removeLocalMessage,
   }) : _signedEventRelay = signedEventRelay,
        _fetchMembers = fetchMembers,
-       _readUserCache = readUserCache;
+       _readUserCache = readUserCache,
+       _addLocalMessage = addLocalMessage,
+       _completeLocalMessage = completeLocalMessage,
+       _removeLocalMessage = removeLocalMessage;
 
   /// Send a text message to a channel.
   ///
@@ -58,11 +69,24 @@ class SendMessage {
       ...mediaTags,
     ];
 
-    await _signedEventRelay.submit(
+    NostrEvent? localMessage;
-      kind: EventKind.streamMessage,
+    try {
-      content: content,
+      await _signedEventRelay.submit(
-      tags: tags,
+        kind: EventKind.streamMessage,
-    );
+        content: content,
+        tags: tags,
+        onSigned: (event) {
+          localMessage = event;
+          _addLocalMessage(channelId, event);
+        },
+      );
+      final event = localMessage;
+      if (event != null) _completeLocalMessage(channelId, event.id);
+    } catch (_) {
+      final event = localMessage;
+      if (event != null) _removeLocalMessage(channelId, event.id);
+      rethrow;
+    }
   }
 
   /// Resolve @mentions to pubkeys, scoped to channel members.
@@ -146,5 +170,14 @@ final sendMessageProvider = Provider<SendMessage>((ref) {
     fetchMembers: (channelId) =>
         ref.read(channelMembersProvider(channelId).future),
     readUserCache: () => ref.read(userCacheProvider),
+    addLocalMessage: (channelId, event) => ref
+        .read(channelMessagesProvider(channelId).notifier)
+        .addLocalMessage(event),
+    completeLocalMessage: (channelId, eventId) => ref
+        .read(channelMessagesProvider(channelId).notifier)
+        .completeLocalMessage(eventId),
+    removeLocalMessage: (channelId, eventId) => ref
+        .read(channelMessagesProvider(channelId).notifier)
+        .removeLocalMessage(eventId),
   );
 });

mobile/lib/features/channels/thread_detail_page.dart

@@ -54,7 +54,7 @@ class ThreadDetailPage extends HookConsumerWidget {
   @override
   Widget build(BuildContext context, WidgetRef ref) {
     final repliesState = ref.watch(
-      threadRepliesProvider(
+      threadRepliesWithLocalProvider(
         ThreadRepliesArgs(channelId: channelId, rootId: threadHead.id),
       ),
     );

mobile/lib/features/channels/thread_replies_provider.dart

@@ -1,6 +1,9 @@
+import 'dart:async';
+
 import 'package:hooks_riverpod/hooks_riverpod.dart';
 
 import '../../shared/relay/relay.dart';
+import 'pending_local_messages_provider.dart';
 
 class ThreadRepliesArgs {
   final String channelId;
@@ -64,3 +67,77 @@ NostrFilter _threadRepliesFilter(
     },
   );
 }
+
+class ThreadLocalRepliesNotifier extends Notifier<List<NostrEvent>> {
+  final ThreadRepliesArgs args;
+
+  ThreadLocalRepliesNotifier(this.args);
+
+  @override
+  List<NostrEvent> build() => const [];
+
+  void add(NostrEvent event) {
+    state = _mergeReplies(state, [event]);
+  }
+
+  void remove(String eventId) {
+    state = state.where((event) => event.id != eventId).toList();
+  }
+
+  void confirm(Set<String> eventIds) {
+    if (!state.any((event) => eventIds.contains(event.id))) return;
+    state = state.where((event) => !eventIds.contains(event.id)).toList();
+  }
+}
+
+final threadLocalRepliesProvider =
+    NotifierProvider.family<
+      ThreadLocalRepliesNotifier,
+      List<NostrEvent>,
+      ThreadRepliesArgs
+    >(ThreadLocalRepliesNotifier.new);
+
+/// Relay-backed replies merged with signed local replies that are still
+/// waiting for acknowledgement.
+final threadRepliesWithLocalProvider =
+    Provider.family<AsyncValue<List<NostrEvent>>, ThreadRepliesArgs>((
+      ref,
+      args,
+    ) {
+      final relayReplies = ref.watch(threadRepliesProvider(args));
+      final localReplies = ref.watch(threadLocalRepliesProvider(args));
+      final authoritative = relayReplies.value;
+      if (authoritative != null && localReplies.isNotEmpty) {
+        final authoritativeIds = authoritative.map((event) => event.id).toSet();
+        if (localReplies.any((event) => authoritativeIds.contains(event.id))) {
+          Future.microtask(() {
+            ref
+                .read(threadLocalRepliesProvider(args).notifier)
+                .confirm(authoritativeIds);
+            ref
+                .read(pendingLocalMessagesProvider(args.channelId).notifier)
+                .confirm(authoritativeIds);
+          });
+        }
+      }
+      if (localReplies.isEmpty) return relayReplies;
+      return relayReplies.when(
+        data: (events) => AsyncData(_mergeReplies(events, localReplies)),
+        loading: () => AsyncData(localReplies),
+        error: (error, stackTrace) => AsyncData(localReplies),
+      );
+    });
+
+List<NostrEvent> _mergeReplies(
+  Iterable<NostrEvent> first,
+  Iterable<NostrEvent> second,
+) {
+  final byId = <String, NostrEvent>{};
+  for (final event in [...first, ...second]) {
+    byId[event.id] = event;
+  }
+  return byId.values.toList()..sort((a, b) {
+    final createdAt = a.createdAt.compareTo(b.createdAt);
+    return createdAt != 0 ? createdAt : a.id.compareTo(b.id);
+  });
+}

mobile/lib/shared/relay/signed_event_relay.dart

@@ -31,6 +31,7 @@ class SignedEventRelay {
     required String content,
     required List<List<String>> tags,
     int? createdAt,
+    void Function(NostrEvent event)? onSigned,
   }) async {
     final nsec = _nsec;
     if (nsec == null || nsec.isEmpty) {
@@ -52,6 +53,7 @@ class SignedEventRelay {
     );
 
     final nostrEvent = NostrEvent.fromJson(event.toMap());
+    onSigned?.call(nostrEvent);
     return _session.publish(nostrEvent);
   }
 }

mobile/test/features/channels/channel_messages_provider_test.dart

@@ -5,6 +5,8 @@ import 'dart:convert';
 import 'package:flutter_test/flutter_test.dart';
 import 'package:hooks_riverpod/hooks_riverpod.dart';
 import 'package:buzz/features/channels/channel_messages_provider.dart';
+import 'package:buzz/features/channels/pending_local_messages_provider.dart';
+import 'package:buzz/features/channels/thread_replies_provider.dart';
 import 'package:buzz/shared/relay/relay.dart';
 
 void main() {
@@ -130,6 +132,369 @@ void main() {
     },
   );
 
+  test(
+    'adds and rolls back a local message in the websocket timeline',
+    () async {
+      final relaySession = _RecordingRelaySessionNotifier();
+      final container = _buildContainer(relaySession);
+      addTearDown(container.dispose);
+
+      container.read(channelMessagesProvider(_channelId));
+      await relaySession.subscribed;
+      final notifier = container.read(
+        channelMessagesProvider(_channelId).notifier,
+      );
+
+      notifier.addLocalMessage(_event(id: 'local', createdAt: 20));
+      expect(
+        container
+            .read(channelMessagesProvider(_channelId))
+            .value
+            ?.map((event) => event.id),
+        ['local'],
+      );
+
+      relaySession.completeHistory([_event(id: 'history', createdAt: 10)]);
+      await _pumpEventQueue();
+
+      // The initial history merge must retain a local row even if the relay's
+      // history snapshot was taken before that outgoing event was durable.
+      expect(
+        container
+            .read(channelMessagesProvider(_channelId))
+            .value
+            ?.map((event) => event.id),
+        ['history', 'local'],
+      );
+
+      notifier.removeLocalMessage('local');
+      expect(
+        container
+            .read(channelMessagesProvider(_channelId))
+            .value
+            ?.map((event) => event.id),
+        ['history'],
+      );
+    },
+  );
+
+  test(
+    'legacy websocket echo retires ownership without duplicating the row',
+    () async {
+      final relaySession = _RecordingRelaySessionNotifier();
+      final container = _buildContainer(relaySession);
+      addTearDown(container.dispose);
+
+      container.read(channelMessagesProvider(_channelId));
+      await relaySession.subscribed;
+      final notifier = container.read(
+        channelMessagesProvider(_channelId).notifier,
+      );
+      final local = _event(id: 'local', createdAt: 20);
+      notifier.addLocalMessage(local);
+
+      relaySession.emit(local);
+      await _pumpEventQueue();
+
+      expect(container.read(pendingLocalMessagesProvider(_channelId)), isEmpty);
+      expect(
+        container
+            .read(channelMessagesProvider(_channelId))
+            .value
+            ?.map((event) => event.id),
+        ['local'],
+      );
+    },
+  );
+
+  test('adds and rolls back a local message in the channel window', () async {
+    final relaySession = _RecordingRelaySessionNotifier(
+      queryResults: [
+        [_event(id: 'history', createdAt: 10), _bounds()],
+      ],
+    );
+    final container = _buildContainer(relaySession);
+    addTearDown(container.dispose);
+
+    container.read(channelMessagesProvider(_channelId));
+    await relaySession.subscribed;
+    await _pumpEventQueue();
+    final notifier = container.read(
+      channelMessagesProvider(_channelId).notifier,
+    );
+
+    notifier.addLocalMessage(_event(id: 'local', createdAt: 20));
+    expect(
+      container
+          .read(channelMessagesProvider(_channelId))
+          .value
+          ?.map((event) => event.id),
+      ['history', 'local'],
+    );
+
+    notifier.removeLocalMessage('local');
+    expect(
+      container
+          .read(channelMessagesProvider(_channelId))
+          .value
+          ?.map((event) => event.id),
+      ['history'],
+    );
+  });
+
+  test('reconnect hydration cannot retain a rolled-back local row', () async {
+    final relaySession = _RecordingRelaySessionNotifier(
+      queryResults: [
+        [_event(id: 'history', createdAt: 10), _bounds()],
+        [_event(id: 'history', createdAt: 10), _bounds()],
+      ],
+    );
+    final container = _buildContainer(relaySession);
+    addTearDown(container.dispose);
+
+    container.read(channelMessagesProvider(_channelId));
+    await relaySession.subscribed;
+    await _pumpEventQueue();
+    final notifier = container.read(
+      channelMessagesProvider(_channelId).notifier,
+    );
+    notifier.addLocalMessage(_event(id: 'local', createdAt: 20));
+
+    relaySession.setConnected(false);
+    await _pumpEventQueue();
+    relaySession.setConnected(true);
+    await _pumpEventQueue();
+    expect(
+      container
+          .read(channelMessagesProvider(_channelId))
+          .value
+          ?.map((event) => event.id),
+      ['history', 'local'],
+    );
+
+    notifier.removeLocalMessage('local');
+    expect(
+      container
+          .read(channelMessagesProvider(_channelId))
+          .value
+          ?.map((event) => event.id),
+      ['history'],
+    );
+  });
+
+  test(
+    'thread replies are inserted, deduped, and rolled back locally',
+    () async {
+      final relaySession = _RecordingRelaySessionNotifier(
+        queryResults: [
+          [_event(id: 'history', createdAt: 10), _bounds()],
+          <NostrEvent>[],
+          [
+            _event(
+              id: 'reply',
+              createdAt: 20,
+              extraTags: const [
+                ['e', 'root', '', 'reply'],
+              ],
+            ),
+          ],
+        ],
+      );
+      final container = _buildContainer(relaySession);
+      addTearDown(container.dispose);
+
+      container.read(channelMessagesProvider(_channelId));
+      await relaySession.subscribed;
+      await _pumpEventQueue();
+      const args = ThreadRepliesArgs(channelId: _channelId, rootId: 'root');
+      container.read(threadRepliesWithLocalProvider(args));
+      await _pumpEventQueue();
+      final notifier = container.read(
+        channelMessagesProvider(_channelId).notifier,
+      );
+      final reply = _event(
+        id: 'reply',
+        createdAt: 20,
+        extraTags: const [
+          ['e', 'root', '', 'reply'],
+        ],
+      );
+
+      notifier.addLocalMessage(reply);
+      expect(
+        container
+            .read(threadRepliesWithLocalProvider(args))
+            .value
+            ?.map((event) => event.id),
+        ['reply'],
+      );
+      expect(
+        container
+            .read(channelMessagesProvider(_channelId))
+            .value
+            ?.map((event) => event.id),
+        ['history'],
+      );
+
+      relaySession.emit(reply);
+      await container.read(threadRepliesProvider(args).future);
+      container.read(threadRepliesWithLocalProvider(args));
+      await _pumpEventQueue();
+      expect(
+        container
+            .read(threadRepliesWithLocalProvider(args))
+            .value
+            ?.map((event) => event.id),
+        ['reply'],
+      );
+      expect(container.read(threadLocalRepliesProvider(args)), isEmpty);
+      expect(container.read(pendingLocalMessagesProvider(_channelId)), isEmpty);
+
+      final rejected = _event(
+        id: 'rejected',
+        createdAt: 21,
+        extraTags: const [
+          ['e', 'root', '', 'reply'],
+        ],
+      );
+      notifier.addLocalMessage(rejected);
+      notifier.removeLocalMessage('rejected');
+      expect(
+        container
+            .read(threadRepliesWithLocalProvider(args))
+            .value
+            ?.map((event) => event.id),
+        ['reply'],
+      );
+    },
+  );
+
+  test(
+    'thread live echo keeps ownership until the authoritative refetch succeeds',
+    () async {
+      final relaySession = _RecordingRelaySessionNotifier(
+        queryResults: [
+          [_event(id: 'history', createdAt: 10), _bounds()],
+          <NostrEvent>[],
+          Exception('thread refetch failed'),
+        ],
+      );
+      final container = _buildContainer(relaySession);
+      addTearDown(container.dispose);
+
+      container.read(channelMessagesProvider(_channelId));
+      await relaySession.subscribed;
+      await _pumpEventQueue();
+      const args = ThreadRepliesArgs(channelId: _channelId, rootId: 'root');
+      container.read(threadRepliesWithLocalProvider(args));
+      await _pumpEventQueue();
+      final notifier = container.read(
+        channelMessagesProvider(_channelId).notifier,
+      );
+      final reply = _event(
+        id: 'reply',
+        createdAt: 20,
+        extraTags: const [
+          ['e', 'root', '', 'reply'],
+        ],
+      );
+      notifier.addLocalMessage(reply);
+
+      relaySession.emit(reply);
+      await _pumpEventQueue();
+
+      expect(container.read(pendingLocalMessagesProvider(_channelId)).keys, [
+        'reply',
+      ]);
+      expect(
+        container
+            .read(threadLocalRepliesProvider(args))
+            .map((event) => event.id),
+        ['reply'],
+      );
+      expect(
+        container
+            .read(threadRepliesWithLocalProvider(args))
+            .value
+            ?.map((event) => event.id),
+        ['reply'],
+      );
+    },
+  );
+
+  test(
+    'successful never-echoed send releases ownership but keeps its row across reconnect',
+    () async {
+      final relaySession = _RecordingRelaySessionNotifier(
+        queryResults: [
+          [_event(id: 'history', createdAt: 10), _bounds()],
+          [_event(id: 'history', createdAt: 10), _bounds()],
+        ],
+      );
+      final container = _buildContainer(relaySession);
+      addTearDown(container.dispose);
+
+      container.read(channelMessagesProvider(_channelId));
+      await relaySession.subscribed;
+      await _pumpEventQueue();
+      final notifier = container.read(
+        channelMessagesProvider(_channelId).notifier,
+      );
+      notifier.addLocalMessage(_event(id: 'local', createdAt: 20));
+      notifier.completeLocalMessage('local');
+
+      expect(container.read(pendingLocalMessagesProvider(_channelId)), isEmpty);
+      relaySession.setConnected(false);
+      await _pumpEventQueue();
+      relaySession.setConnected(true);
+      await _pumpEventQueue();
+
+      expect(container.read(pendingLocalMessagesProvider(_channelId)), isEmpty);
+      expect(
+        container
+            .read(channelMessagesProvider(_channelId))
+            .value
+            ?.map((event) => event.id),
+        ['history', 'local'],
+      );
+    },
+  );
+
+  test(
+    'window dedupes echoes and orders rapid equal-time local sends',
+    () async {
+      final relaySession = _RecordingRelaySessionNotifier(
+        queryResults: [
+          [_event(id: 'history', createdAt: 10), _bounds()],
+        ],
+      );
+      final container = _buildContainer(relaySession);
+      addTearDown(container.dispose);
+
+      container.read(channelMessagesProvider(_channelId));
+      await relaySession.subscribed;
+      await _pumpEventQueue();
+      final notifier = container.read(
+        channelMessagesProvider(_channelId).notifier,
+      );
+      notifier.addLocalMessage(_event(id: 'z-local', createdAt: 20));
+      notifier.addLocalMessage(_event(id: 'a-local', createdAt: 20));
+      relaySession.emit(_event(id: 'z-local', createdAt: 20));
+      await _pumpEventQueue();
+
+      expect(container.read(pendingLocalMessagesProvider(_channelId)).keys, [
+        'a-local',
+      ]);
+      expect(
+        container
+            .read(channelMessagesProvider(_channelId))
+            .value
+            ?.map((event) => event.id),
+        ['history', 'z-local', 'a-local'],
+      );
+    },
+  );
+
   test('window pagination failures return false without exhausting', () async {
     final relaySession = _RecordingRelaySessionNotifier(
       queryResults: [
@@ -178,14 +543,19 @@ ProviderContainer _buildContainer(_RecordingRelaySessionNotifier relaySession) {
   );
 }
 
-NostrEvent _event({required String id, required int createdAt}) {
+NostrEvent _event({
+  required String id,
+  required int createdAt,
+  List<List<String>> extraTags = const [],
+}) {
   return NostrEvent(
     id: id,
     pubkey: 'alice',
     createdAt: createdAt,
     kind: EventKind.streamMessageV2,
-    tags: const [
+    tags: [
       ['h', _channelId],
+      ...extraTags,
     ],
     content: id,
     sig: 'sig',
@@ -243,6 +613,12 @@ class _RecordingRelaySessionNotifier extends RelaySessionNotifier {
   @override
   SessionState build() => const SessionState(status: SessionStatus.connected);
 
+  void setConnected(bool connected) {
+    state = SessionState(
+      status: connected ? SessionStatus.connected : SessionStatus.disconnected,
+    );
+  }
+
   @override
   Future<List<NostrEvent>> queryRelay(
     List<NostrFilter> filters, {

mobile/test/features/channels/read_state/read_state_manager_test.dart

@@ -181,6 +181,7 @@ class _FakeSignedEventRelay implements SignedEventRelay {
     required String content,
     required List<List<String>> tags,
     int? createdAt,
+    void Function(NostrEvent event)? onSigned,
   }) async {
     submitted.complete(_SubmittedEvent(kind: kind, tags: tags));
     return _stubAckEvent();
@@ -199,6 +200,7 @@ class _UnsupportedKindSignedEventRelay implements SignedEventRelay {
     required String content,
     required List<List<String>> tags,
     int? createdAt,
+    void Function(NostrEvent event)? onSigned,
   }) async {
     submitCount++;
     throw Exception('restricted: unknown event kind');
@@ -217,6 +219,7 @@ class _MissingScopeSignedEventRelay implements SignedEventRelay {
     required String content,
     required List<List<String>> tags,
     int? createdAt,
+    void Function(NostrEvent event)? onSigned,
   }) async {
     submitCount++;
     throw Exception('missing users:write');

mobile/test/features/channels/send_message_provider_test.dart

@@ -0,0 +1,96 @@
+import 'dart:async';
+
+import 'package:flutter_test/flutter_test.dart';
+import 'package:nostr/nostr.dart' as nostr;
+import 'package:buzz/features/channels/send_message_provider.dart';
+import 'package:buzz/shared/relay/relay.dart';
+
+void main() {
+  test(
+    'adds the signed message locally before relay acknowledgement',
+    () async {
+      final session = _PendingPublishRelaySession();
+      final localMessages = <NostrEvent>[];
+      final removedIds = <String>[];
+      final completedIds = <String>[];
+      final send = SendMessage(
+        signedEventRelay: SignedEventRelay(
+          session: session,
+          nsec: nostr.Keys.generate().nsec,
+        ),
+        fetchMembers: (_) async => const [],
+        readUserCache: () => const {},
+        addLocalMessage: (_, event) => localMessages.add(event),
+        completeLocalMessage: (_, eventId) => completedIds.add(eventId),
+        removeLocalMessage: (_, eventId) => removedIds.add(eventId),
+      );
+
+      final result = send(channelId: _channelId, content: 'hello');
+      await session.published;
+
+      expect(localMessages, hasLength(1));
+      expect(localMessages.single.id, session.event.id);
+      expect(localMessages.single.content, 'hello');
+      expect(localMessages.single.channelId, _channelId);
+      expect(removedIds, isEmpty);
+
+      session.accept();
+      await result;
+      expect(completedIds, [localMessages.single.id]);
+      expect(removedIds, isEmpty);
+    },
+  );
+
+  test('rolls back the signed local message when publish fails', () async {
+    final session = _PendingPublishRelaySession();
+    final localMessages = <NostrEvent>[];
+    final completedIds = <String>[];
+    final removedIds = <String>[];
+    final send = SendMessage(
+      signedEventRelay: SignedEventRelay(
+        session: session,
+        nsec: nostr.Keys.generate().nsec,
+      ),
+      fetchMembers: (_) async => const [],
+      readUserCache: () => const {},
+      addLocalMessage: (_, event) => localMessages.add(event),
+      completeLocalMessage: (_, eventId) => completedIds.add(eventId),
+      removeLocalMessage: (_, eventId) => removedIds.add(eventId),
+    );
+
+    final result = send(channelId: _channelId, content: 'hello');
+    await session.published;
+    session.reject();
+
+    await expectLater(result, throwsException);
+    expect(completedIds, isEmpty);
+    expect(removedIds, [localMessages.single.id]);
+  });
+}
+
+const _channelId = '11111111-1111-4111-8111-111111111111';
+
+class _PendingPublishRelaySession extends RelaySessionNotifier {
+  final Completer<NostrEvent> _result = Completer<NostrEvent>();
+  final Completer<void> _published = Completer<void>();
+  late NostrEvent event;
+
+  Future<void> get published => _published.future;
+
+  @override
+  SessionState build() => const SessionState(status: SessionStatus.connected);
+
+  @override
+  Future<NostrEvent> publish(
+    NostrEvent event, {
+    Duration timeout = const Duration(seconds: 8),
+  }) {
+    this.event = event;
+    _published.complete();
+    return _result.future;
+  }
+
+  void accept() => _result.complete(event);
+
+  void reject() => _result.completeError(Exception('relay rejected event'));
+}