Verify mobile relay events on receipt

Co-authored-by: Codex <noreply@openai.com>
Signed-off-by: jm <jm@squareup.com>

mobile/lib/shared/relay/relay_event_verifier.dart

@@ -0,0 +1,236 @@
+part of 'relay_session.dart';
+
+const _maxPendingRelayEvents = 2048;
+const _relayEventVerificationBatchSize = 64;
+// The relay caps frames at 512 KiB and event content at 256 KiB. This is a
+// conservative upper bound for JSON-encoded bytes: a Dart UTF-16 code unit can
+// expand to six bytes when JSON escaping is required.
+const _maxRelayEventEncodedByteUpperBound = 4 * 1024 * 1024;
+const _maxPendingRelayEventEncodedByteUpperBound = 32 * 1024 * 1024;
+const _maxRelayEventNestingDepth = 64;
+
+/// Serializes relay-event verification through one bounded background worker.
+class _RelayEventVerifier {
+  _RelayEventVerifier({
+    required Future<List<Map<String, dynamic>?>> Function(List<Object?>)
+    verifyBatch,
+  }) : _verifyBatch = verifyBatch;
+
+  final Future<List<Map<String, dynamic>?>> Function(List<Object?>)
+  _verifyBatch;
+  final Queue<_PendingRelayEventVerification> _pending = Queue();
+  List<_PendingRelayEventVerification> _active = const [];
+  Timer? _drainTimer;
+  var _pendingCount = 0;
+  var _pendingEncodedByteUpperBound = 0;
+  var _draining = false;
+  var _closed = false;
+
+  Future<NostrEvent?> verify(Object? eventJson) async {
+    final verified = await verifyAll([eventJson]);
+    return verified.single;
+  }
+
+  Future<List<NostrEvent?>> verifyAll(List<Object?> eventJsons) {
+    if (_closed) {
+      return Future.error(StateError('Relay event verifier is closed'));
+    }
+    final encodedByteUpperBounds = <int?>[];
+    var addedCount = 0;
+    var addedEncodedByteUpperBound = 0;
+    for (final eventJson in eventJsons) {
+      final encodedByteUpperBound = _relayEventEncodedByteUpperBound(eventJson);
+      encodedByteUpperBounds.add(encodedByteUpperBound);
+      if (encodedByteUpperBound != null) {
+        addedCount++;
+        addedEncodedByteUpperBound += encodedByteUpperBound;
+      }
+    }
+    if (addedCount > _maxPendingRelayEvents - _pendingCount ||
+        addedEncodedByteUpperBound >
+            _maxPendingRelayEventEncodedByteUpperBound -
+                _pendingEncodedByteUpperBound) {
+      return Future.error(StateError('Relay event verification queue is full'));
+    }
+
+    final futures = <Future<NostrEvent?>>[];
+    for (var i = 0; i < eventJsons.length; i++) {
+      final encodedByteUpperBound = encodedByteUpperBounds[i];
+      if (encodedByteUpperBound == null) {
+        futures.add(Future.value());
+        continue;
+      }
+      final pending = _PendingRelayEventVerification(
+        eventJsons[i],
+        encodedByteUpperBound,
+      );
+      _pending.add(pending);
+      _pendingCount++;
+      _pendingEncodedByteUpperBound += encodedByteUpperBound;
+      futures.add(pending.completer.future);
+    }
+    _scheduleDrain();
+    return Future.wait(futures);
+  }
+
+  void close() {
+    if (_closed) return;
+    _closed = true;
+    _drainTimer?.cancel();
+    _drainTimer = null;
+    final error = StateError('Relay event verifier is closed');
+    for (final pending in [..._active, ..._pending]) {
+      _completeError(pending, error, StackTrace.current);
+    }
+    _pending.clear();
+  }
+
+  void _scheduleDrain() {
+    if (_closed || _draining || _drainTimer != null || _pending.isEmpty) {
+      return;
+    }
+    // A zero-duration timer coalesces adjacent WebSocket frames into batches.
+    _drainTimer = Timer(Duration.zero, () {
+      _drainTimer = null;
+      unawaited(_drain());
+    });
+  }
+
+  Future<void> _drain() async {
+    if (_closed || _draining) return;
+    _draining = true;
+    try {
+      while (!_closed && _pending.isNotEmpty) {
+        final batch = <_PendingRelayEventVerification>[];
+        while (batch.length < _relayEventVerificationBatchSize &&
+            _pending.isNotEmpty) {
+          batch.add(_pending.removeFirst());
+        }
+        _active = batch;
+
+        try {
+          final verified = await _verifyBatch([
+            for (final pending in batch) pending.eventJson,
+          ]);
+          if (verified.length != batch.length) {
+            throw StateError(
+              'Relay event verifier returned ${verified.length} results '
+              'for ${batch.length} events',
+            );
+          }
+          for (var i = 0; i < batch.length; i++) {
+            final eventJson = verified[i];
+            _complete(
+              batch[i],
+              eventJson == null ? null : NostrEvent.fromJson(eventJson),
+            );
+          }
+        } catch (error, stackTrace) {
+          for (final pending in batch) {
+            _completeError(pending, error, stackTrace);
+          }
+        } finally {
+          _active = const [];
+        }
+      }
+    } finally {
+      _draining = false;
+      _scheduleDrain();
+    }
+  }
+
+  void _complete(_PendingRelayEventVerification pending, NostrEvent? event) {
+    if (pending.completer.isCompleted) return;
+    _pendingCount--;
+    _pendingEncodedByteUpperBound -= pending.encodedByteUpperBound;
+    pending.completer.complete(event);
+  }
+
+  void _completeError(
+    _PendingRelayEventVerification pending,
+    Object error,
+    StackTrace stackTrace,
+  ) {
+    if (pending.completer.isCompleted) return;
+    _pendingCount--;
+    _pendingEncodedByteUpperBound -= pending.encodedByteUpperBound;
+    pending.completer.completeError(error, stackTrace);
+  }
+}
+
+class _PendingRelayEventVerification {
+  _PendingRelayEventVerification(this.eventJson, this.encodedByteUpperBound);
+
+  final Object? eventJson;
+  final int encodedByteUpperBound;
+  final Completer<NostrEvent?> completer = Completer();
+}
+
+int? _relayEventEncodedByteUpperBound(Object? value) {
+  return _encodedByteUpperBound(value, _maxRelayEventEncodedByteUpperBound, 0);
+}
+
+int? _encodedByteUpperBound(Object? value, int remaining, int depth) {
+  if (remaining < 0 || depth > _maxRelayEventNestingDepth) return null;
+  if (value == null || value is bool || value is num) {
+    return remaining >= 32 ? 32 : null;
+  }
+  if (value is String) {
+    final bytes = 2 + value.length * 6;
+    return bytes <= remaining ? bytes : null;
+  }
+  if (value is List) {
+    var total = 2;
+    for (final element in value) {
+      final elementBytes = _encodedByteUpperBound(
+        element,
+        remaining - total - 1,
+        depth + 1,
+      );
+      if (elementBytes == null) return null;
+      total += elementBytes + 1;
+    }
+    return total <= remaining ? total : null;
+  }
+  if (value is Map) {
+    var total = 2;
+    for (final entry in value.entries) {
+      if (entry.key is! String) return null;
+      final keyBytes = _encodedByteUpperBound(
+        entry.key,
+        remaining - total - 1,
+        depth + 1,
+      );
+      if (keyBytes == null) return null;
+      total += keyBytes + 1;
+      final valueBytes = _encodedByteUpperBound(
+        entry.value,
+        remaining - total - 1,
+        depth + 1,
+      );
+      if (valueBytes == null) return null;
+      total += valueBytes + 1;
+    }
+    return total <= remaining ? total : null;
+  }
+  return null;
+}
+
+Future<List<Map<String, dynamic>?>> _verifyRelayEventBatchInBackground(
+  List<Object?> eventJsons,
+) => compute(_verifyRelayEventBatch, eventJsons);
+
+List<Map<String, dynamic>?> _verifyRelayEventBatch(List<Object?> eventJsons) {
+  return [for (final eventJson in eventJsons) _verifyRelayEvent(eventJson)];
+}
+
+Map<String, dynamic>? _verifyRelayEvent(Object? eventJson) {
+  try {
+    // Event.fromJson verifies the canonical seven-field ID and BIP-340
+    // signature. Its Schnorr boundary strictly decodes the complete key,
+    // message, and signature widths before invoking bip340 verification.
+    return nostr.Event.fromJson(jsonEncode(eventJson)).toMap();
+  } catch (_) {
+    return null;
+  }
+}

mobile/lib/shared/relay/relay_session.dart

@@ -1,4 +1,5 @@
 import 'dart:async';
+import 'dart:collection';
 import 'dart:convert';
 import 'dart:math';
 
@@ -24,11 +25,18 @@ import 'relay_socket.dart';
 export 'relay_session_types.dart';
 
 part 'relay_session_auth.dart';
+part 'relay_event_verifier.dart';
+part 'relay_session_event_admission.dart';
 
 class _HistorySubscription {
   final List<NostrEvent> events = [];
   final Completer<List<NostrEvent>> completer;
   final Timer timeout;
+  Future<void> verificationTail = Future.value();
+  int pendingVerifications = 0;
+  int receivedEventFrames = 0;
+  Object? verificationError;
+  StackTrace? verificationStackTrace;
 
   _HistorySubscription({required this.completer, required this.timeout});
 }
@@ -42,6 +50,11 @@ class _LiveSubscription {
   int? lastSeenCreatedAt;
   int closedRetryAttempt = 0;
   Timer? closedRetryTimer;
+  Future<void> verificationTail = Future.value();
+  int pendingVerifications = 0;
+  int receivedEventFrames = 0;
+  int validEventFrames = 0;
+  int verificationEpoch = 0;
 
   _LiveSubscription({
     required this.filter,
@@ -73,6 +86,12 @@ class _BufferedEvent {
 }
 
 class RelaySessionNotifier extends Notifier<SessionState> {
+  static const _maxQueryResponseBytes = 8 * 1024 * 1024;
+
+  /// Creates a relay session.
+  ///
+  /// [verifyEventBatch] is a test seam for controlling asynchronous event
+  /// admission. Production callers must use the default background verifier.
   RelaySessionNotifier({
     http.Client? httpClient,
     http.Client Function()? httpClientFactory,
@@ -81,6 +100,8 @@ class RelaySessionNotifier extends Notifier<SessionState> {
     RelayRateLimitGate? rateLimitGate,
     RelayTimerFactory retryTimerFactory = Timer.new,
     Future<void> Function(Duration) replayDelay = Future.delayed,
+    Future<List<Map<String, dynamic>?>> Function(List<Object?>)?
+    verifyEventBatch,
   }) : _httpQueryClient = RelayHttpQueryClient(
          client: httpClient,
          clientFactory: httpClientFactory,
@@ -89,7 +110,9 @@ class RelaySessionNotifier extends Notifier<SessionState> {
        _now = now ?? DateTime.now,
        _rateLimitGate = rateLimitGate ?? RelayRateLimitGate(),
        _retryTimerFactory = retryTimerFactory,
-       _replayDelay = replayDelay;
+       _replayDelay = replayDelay,
+       _verifyEventBatch =
+           verifyEventBatch ?? _verifyRelayEventBatchInBackground;
 
   final RelayHttpQueryClient _httpQueryClient;
   final RelaySocketFactory _socketFactory;
@@ -97,6 +120,8 @@ class RelaySessionNotifier extends Notifier<SessionState> {
   final RelayRateLimitGate _rateLimitGate;
   final RelayTimerFactory _retryTimerFactory;
   final Future<void> Function(Duration) _replayDelay;
+  final Future<List<Map<String, dynamic>?>> Function(List<Object?>)
+  _verifyEventBatch;
 
   static const _baseReconnectDelayMs = 1000;
   static const _maxReconnectDelayMs = 30000;
@@ -130,6 +155,10 @@ class RelaySessionNotifier extends Notifier<SessionState> {
   final Map<Object, Future<void> Function()> _beforePauseCallbacks = {};
   bool _socketConnected = false;
   bool _closedRetryReplayScheduled = false;
+  _RelayEventVerifier? _eventVerifier;
+
+  _RelayEventVerifier get _verifier =>
+      _eventVerifier ??= _RelayEventVerifier(verifyBatch: _verifyEventBatch);
 
   @override
   SessionState build() {
@@ -190,22 +219,30 @@ class RelaySessionNotifier extends Notifier<SessionState> {
       _activateRateLimitGateFromHttpError(response.body);
       throw RelayException(response.statusCode, response.body);
     }
+    if (response.bodyBytes.length > _maxQueryResponseBytes) {
+      throw const FormatException('relay query response exceeds size limit');
+    }
     final decoded = jsonDecode(response.body);
     if (decoded is! List) {
       throw const FormatException('relay returned malformed query response');
     }
-    try {
-      return [
-        for (final eventJson in decoded)
-          if (eventJson is Map<String, dynamic>)
-            NostrEvent.fromJson(eventJson)
-          else
-            throw const FormatException('relay returned malformed query event'),
-      ];
-    } catch (error) {
-      if (error is FormatException) rethrow;
-      throw FormatException('relay returned malformed query event: $error');
+    final verified = await _verifier.verifyAll(decoded.cast<Object?>());
+    if (_disposed || _ageRestricted || generation != _contextGeneration) {
+      throw StateError('Relay query belongs to a retired session');
     }
+    final events = [
+      for (final event in verified)
+        if (event != null) event,
+    ];
+    // A mixed response can still contribute its independently verified events.
+    // A non-empty response with no valid events is not authoritative absence:
+    // surface it as a failure so callers cannot wipe caches with forged data.
+    if (decoded.isNotEmpty && events.isEmpty) {
+      throw const FormatException(
+        'relay query returned no valid signed events',
+      );
+    }
+    return events;
   }
 
   void _activateRateLimitGateFromHttpError(String body) {
@@ -373,7 +410,10 @@ class RelaySessionNotifier extends Notifier<SessionState> {
   void debugDispose() => _dispose();
 
   @visibleForTesting
-  void debugSupersedeConnection() => _connectionGeneration++;
+  void debugSupersedeConnection() {
+    _connectionGeneration++;
+    _retireLiveVerification();
+  }
 
   @visibleForTesting
   void debugHandleDisconnected([Object? error]) {
@@ -417,6 +457,7 @@ class RelaySessionNotifier extends Notifier<SessionState> {
   /// Force a reconnect (e.g., returning from background).
   Future<void> reconnect() async {
     _socketConnected = false;
+    _retireLiveVerification();
     await _socket?.disconnect();
     _reconnectDelayMs = _baseReconnectDelayMs;
     final config = ref.read(relayConfigProvider);
@@ -452,6 +493,7 @@ class RelaySessionNotifier extends Notifier<SessionState> {
   void _pauseNow() {
     _paused = true;
     _socketConnected = false;
+    _retireLiveVerification();
     _reconnectTimer?.cancel();
     _cancelAllHistory(Exception('App moved to background'));
     _rejectAllPending(Exception('App moved to background'));
@@ -487,6 +529,7 @@ class RelaySessionNotifier extends Notifier<SessionState> {
     if (_disposed || _ageRestricted) return;
 
     final generation = ++_connectionGeneration;
+    _retireLiveVerification();
     state = SessionState(
       status: _hasConnectedOnce
           ? SessionStatus.reconnecting
@@ -521,6 +564,7 @@ class RelaySessionNotifier extends Notifier<SessionState> {
   void _handleDisconnected(int generation, Object? error) {
     if (_disposed || generation != _connectionGeneration) return;
     _socketConnected = false;
+    _retireLiveVerification();
     _cancelAllHistory(error);
     _rejectAllPending(error);
     _resetAllClosedRetries();
@@ -633,7 +677,8 @@ class RelaySessionNotifier extends Notifier<SessionState> {
 
   void _handleMessage(List<dynamic> data) {
     if (data.isEmpty) return;
-    final type = data[0] as String;
+    final type = data[0];
+    if (type is! String) return;
 
     switch (type) {
       case 'EVENT':
@@ -647,66 +692,6 @@ class RelaySessionNotifier extends Notifier<SessionState> {
     }
   }
 
-  void _handleEvent(List<dynamic> data) {
-    if (data.length < 3) return;
-    final subId = data[1] as String;
-    final eventJson = data[2] as Map<String, dynamic>;
-    final event = NostrEvent.fromJson(eventJson);
-
-    // History subscriptions accumulate immediately.
-    final historySub = _historySubscriptions[subId];
-    if (historySub != null) {
-      historySub.events.add(event);
-      return;
-    }
-
-    // Live subscriptions get batched. An EVENT proves the stream is active,
-    // but not that a retry replay is complete; only EOSE is that boundary.
-    final liveSub = _liveSubscriptions[subId];
-    if (liveSub != null) {
-      _resetClosedRetry(liveSub);
-      // Track last seen timestamp for reconnect replay.
-      if (liveSub.lastSeenCreatedAt == null ||
-          event.createdAt > liveSub.lastSeenCreatedAt!) {
-        liveSub.lastSeenCreatedAt = event.createdAt;
-      }
-      _eventBuffer.add(_BufferedEvent(subId, event));
-      _scheduleFlush();
-    }
-  }
-
-  void _handleEose(List<dynamic> data) {
-    if (data.length < 2) return;
-    final subId = data[1] as String;
-
-    // History subscription: resolve with collected events.
-    final historySub = _historySubscriptions.remove(subId);
-    if (historySub != null) {
-      historySub.timeout.cancel();
-      if (!historySub.completer.isCompleted) {
-        historySub.completer.complete(historySub.events);
-      }
-      _sendClose(subId);
-      return;
-    }
-
-    // Live subscription: flush replay callbacks before signaling ready. This
-    // ordering matters for retry replays, whose original ready completer has
-    // already been released.
-    final liveSub = _liveSubscriptions[subId];
-    if (liveSub != null) {
-      _resetClosedRetry(liveSub);
-      _flushBufferedEventsNow();
-      liveSub.onStatusChanged?.call(RelaySubscriptionStatus.ready);
-    }
-    if (liveSub != null &&
-        liveSub.readyCompleter != null &&
-        !liveSub.readyCompleter!.isCompleted) {
-      liveSub.readyCompleter!.complete();
-      liveSub.readyCompleter = null;
-    }
-  }
-
   void _handleClosed(List<dynamic> data) {
     if (data.length < 2) return;
     final subId = data[1] as String;
@@ -925,6 +910,7 @@ class RelaySessionNotifier extends Notifier<SessionState> {
   void _removeLiveSubscription(String subId, _LiveSubscription subscription) {
     if (_liveSubscriptions[subId] != subscription) return;
     _liveSubscriptions.remove(subId);
+    _retireLiveSubscriptionVerification(subscription);
     _pendingClosedRetries.remove(subId);
     subscription.closedRetryTimer?.cancel();
     subscription.closedRetryTimer = null;
@@ -994,6 +980,9 @@ class RelaySessionNotifier extends Notifier<SessionState> {
       subscription.closedRetryTimer = null;
     }
     _recentDeliveryKeys.clear();
+    _eventBuffer.clear();
+    _eventVerifier?.close();
+    _eventVerifier = null;
     _socket?.dispose();
     _socket = null;
     _httpQueryClient.close();

mobile/lib/shared/relay/relay_session_event_admission.dart

@@ -0,0 +1,240 @@
+part of 'relay_session.dart';
+
+extension _RelaySessionEventAdmission on RelaySessionNotifier {
+  void _handleEvent(List<dynamic> data) {
+    if (data.length < 3 || data[1] is! String) return;
+    final subId = data[1] as String;
+    final connectionGeneration = _connectionGeneration;
+    final contextGeneration = _contextGeneration;
+
+    final historySub = _historySubscriptions[subId];
+    if (historySub != null) {
+      historySub.receivedEventFrames++;
+      historySub.pendingVerifications++;
+      final verification = _verifier.verify(data[2]);
+      historySub.verificationTail = historySub.verificationTail.then((_) async {
+        try {
+          final event = await verification;
+          if (_historySubscriptions[subId] != historySub ||
+              connectionGeneration != _connectionGeneration ||
+              contextGeneration != _contextGeneration) {
+            return;
+          }
+          if (event != null) historySub.events.add(event);
+        } catch (error, stackTrace) {
+          if (_historySubscriptions[subId] == historySub &&
+              connectionGeneration == _connectionGeneration &&
+              contextGeneration == _contextGeneration) {
+            historySub.verificationError ??= error;
+            historySub.verificationStackTrace ??= stackTrace;
+          }
+        } finally {
+          historySub.pendingVerifications--;
+        }
+      });
+      return;
+    }
+
+    final liveSub = _liveSubscriptions[subId];
+    if (liveSub != null) {
+      final verificationEpoch = liveSub.verificationEpoch;
+      liveSub.receivedEventFrames++;
+      liveSub.pendingVerifications++;
+      final verification = _verifier.verify(data[2]);
+      liveSub.verificationTail = liveSub.verificationTail.then((_) async {
+        try {
+          final event = await verification;
+          if (_liveSubscriptions[subId] != liveSub ||
+              connectionGeneration != _connectionGeneration ||
+              contextGeneration != _contextGeneration ||
+              verificationEpoch != liveSub.verificationEpoch ||
+              event == null) {
+            return;
+          }
+
+          // Only a verified event can prove activity or advance replay state.
+          _resetClosedRetry(liveSub);
+          liveSub.validEventFrames++;
+          if (liveSub.lastSeenCreatedAt == null ||
+              event.createdAt > liveSub.lastSeenCreatedAt!) {
+            liveSub.lastSeenCreatedAt = event.createdAt;
+          }
+          _eventBuffer.add(_BufferedEvent(subId, event));
+          _scheduleFlush();
+        } catch (error, stackTrace) {
+          if (_liveSubscriptions[subId] == liveSub &&
+              connectionGeneration == _connectionGeneration &&
+              contextGeneration == _contextGeneration &&
+              verificationEpoch == liveSub.verificationEpoch) {
+            _failLiveVerification(subId, liveSub, error, stackTrace);
+          }
+        } finally {
+          if (verificationEpoch == liveSub.verificationEpoch) {
+            liveSub.pendingVerifications--;
+          }
+        }
+      });
+    }
+  }
+
+  void _handleEose(List<dynamic> data) {
+    if (data.length < 2 || data[1] is! String) return;
+    final subId = data[1] as String;
+
+    final historySub = _historySubscriptions[subId];
+    if (historySub != null) {
+      if (historySub.pendingVerifications == 0) {
+        _completeHistoryEose(subId, historySub);
+      } else {
+        unawaited(_completeHistoryEoseAfterVerification(subId, historySub));
+      }
+      return;
+    }
+
+    final liveSub = _liveSubscriptions[subId];
+    if (liveSub != null) {
+      final connectionGeneration = _connectionGeneration;
+      final contextGeneration = _contextGeneration;
+      final verificationEpoch = liveSub.verificationEpoch;
+      if (liveSub.pendingVerifications == 0) {
+        _completeLiveEose(
+          subId,
+          liveSub,
+          connectionGeneration,
+          contextGeneration,
+          verificationEpoch,
+        );
+      } else {
+        unawaited(
+          _completeLiveEoseAfterVerification(
+            subId,
+            liveSub,
+            connectionGeneration,
+            contextGeneration,
+            verificationEpoch,
+          ),
+        );
+      }
+    }
+  }
+
+  Future<void> _completeHistoryEoseAfterVerification(
+    String subId,
+    _HistorySubscription subscription,
+  ) async {
+    await subscription.verificationTail;
+    _completeHistoryEose(subId, subscription);
+  }
+
+  void _completeHistoryEose(String subId, _HistorySubscription subscription) {
+    if (_historySubscriptions[subId] != subscription) return;
+    _historySubscriptions.remove(subId);
+    subscription.timeout.cancel();
+    final verificationError = subscription.verificationError;
+    if (!subscription.completer.isCompleted) {
+      if (verificationError != null) {
+        subscription.completer.completeError(
+          verificationError,
+          subscription.verificationStackTrace,
+        );
+      } else if (subscription.receivedEventFrames > 0 &&
+          subscription.events.isEmpty) {
+        subscription.completer.completeError(
+          const FormatException(
+            'relay history returned no valid signed events',
+          ),
+        );
+      } else {
+        subscription.completer.complete(subscription.events);
+      }
+    }
+    _sendClose(subId);
+  }
+
+  Future<void> _completeLiveEoseAfterVerification(
+    String subId,
+    _LiveSubscription subscription,
+    int connectionGeneration,
+    int contextGeneration,
+    int verificationEpoch,
+  ) async {
+    await subscription.verificationTail;
+    _completeLiveEose(
+      subId,
+      subscription,
+      connectionGeneration,
+      contextGeneration,
+      verificationEpoch,
+    );
+  }
+
+  void _completeLiveEose(
+    String subId,
+    _LiveSubscription subscription,
+    int connectionGeneration,
+    int contextGeneration,
+    int verificationEpoch,
+  ) {
+    if (_liveSubscriptions[subId] != subscription ||
+        connectionGeneration != _connectionGeneration ||
+        contextGeneration != _contextGeneration ||
+        verificationEpoch != subscription.verificationEpoch) {
+      return;
+    }
+    if (subscription.receivedEventFrames > 0 &&
+        subscription.validEventFrames == 0) {
+      _failLiveVerification(
+        subId,
+        subscription,
+        const FormatException(
+          'relay subscription returned no valid signed events',
+        ),
+        StackTrace.current,
+      );
+      return;
+    }
+
+    subscription.receivedEventFrames = 0;
+    subscription.validEventFrames = 0;
+    // Flush replay callbacks before signaling ready. This ordering matters for
+    // retry replays, whose original ready completer has already been released.
+    _resetClosedRetry(subscription);
+    _flushBufferedEventsNow();
+    subscription.onStatusChanged?.call(RelaySubscriptionStatus.ready);
+    final readyCompleter = subscription.readyCompleter;
+    if (readyCompleter != null && !readyCompleter.isCompleted) {
+      readyCompleter.complete();
+      subscription.readyCompleter = null;
+    }
+  }
+
+  void _failLiveVerification(
+    String subId,
+    _LiveSubscription subscription,
+    Object error,
+    StackTrace stackTrace,
+  ) {
+    if (_liveSubscriptions[subId] != subscription) return;
+    final readyCompleter = subscription.readyCompleter;
+    if (readyCompleter != null && !readyCompleter.isCompleted) {
+      readyCompleter.completeError(error, stackTrace);
+    } else {
+      subscription.onClosed?.call('event verification failed: $error');
+    }
+    _removeLiveSubscription(subId, subscription);
+  }
+
+  void _retireLiveVerification() {
+    for (final subscription in _liveSubscriptions.values) {
+      _retireLiveSubscriptionVerification(subscription);
+    }
+  }
+
+  void _retireLiveSubscriptionVerification(_LiveSubscription subscription) {
+    subscription.verificationEpoch++;
+    subscription.verificationTail = Future.value();
+    subscription.pendingVerifications = 0;
+    subscription.receivedEventFrames = 0;
+    subscription.validEventFrames = 0;
+  }
+}

mobile/test/shared/relay/relay_event_ingress_integration_test.dart

@@ -0,0 +1,191 @@
+import 'dart:async';
+import 'dart:convert';
+import 'dart:io';
+
+import 'package:flutter_test/flutter_test.dart';
+import 'package:hooks_riverpod/hooks_riverpod.dart';
+import 'package:nostr/nostr.dart' as nostr;
+
+import 'package:buzz/shared/auth/auth_provider.dart';
+import 'package:buzz/shared/relay/nostr_models.dart';
+import 'package:buzz/shared/relay/relay_provider.dart';
+import 'package:buzz/shared/relay/relay_session.dart';
+
+void main() {
+  test('real HTTP and WebSocket ingress admits only verified events', () async {
+    final queryEvent = _signedEvent('query');
+    final historyEvent = _signedEvent('history');
+    final liveEvent = _signedEvent('live');
+    final relay = await _AdversarialRelay.start(
+      queryEvents: [_forgedEvent('forged query'), queryEvent.toJson()],
+      historyEvents: [_forgedEvent('forged history'), historyEvent.toJson()],
+    );
+    addTearDown(relay.close);
+
+    final keys = nostr.Keys.generate();
+    final session = RelaySessionNotifier();
+    final container = ProviderContainer(
+      overrides: [
+        authProvider.overrideWith(() => _AuthenticatedAuthNotifier()),
+        relayConfigProvider.overrideWith(
+          () => _TestRelayConfigNotifier(relay.origin, keys.nsec),
+        ),
+        relaySessionProvider.overrideWith(() => session),
+      ],
+    );
+    addTearDown(container.dispose);
+    await container.read(authProvider.future);
+    final listener = container.listen(relaySessionProvider, (_, _) {});
+    addTearDown(listener.close);
+    await _waitUntil(
+      () =>
+          container.read(relaySessionProvider).status ==
+          SessionStatus.connected,
+    );
+
+    expect(await session.queryRelay(const []), [queryEvent]);
+    expect(await session.fetchHistory(_filter), [historyEvent]);
+
+    final delivered = <NostrEvent>[];
+    final unsubscribe = await session.subscribe(_filter, delivered.add);
+    addTearDown(unsubscribe);
+    await relay.sendLive([_forgedEvent('forged live'), liveEvent.toJson()]);
+    await _waitUntil(() => delivered.isNotEmpty);
+
+    expect(delivered, [liveEvent]);
+  });
+}
+
+class _AdversarialRelay {
+  _AdversarialRelay._(this._server, this.queryEvents, this.historyEvents);
+
+  final HttpServer _server;
+  final List<Object?> queryEvents;
+  final List<Object?> historyEvents;
+  final List<WebSocket> _sockets = [];
+  final Completer<String> _liveSubscription = Completer();
+
+  String get origin => 'http://${_server.address.address}:${_server.port}';
+
+  static Future<_AdversarialRelay> start({
+    required List<Object?> queryEvents,
+    required List<Object?> historyEvents,
+  }) async {
+    final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
+    final relay = _AdversarialRelay._(server, queryEvents, historyEvents);
+    server.listen(relay._handleRequest);
+    return relay;
+  }
+
+  Future<void> _handleRequest(HttpRequest request) async {
+    if (WebSocketTransformer.isUpgradeRequest(request)) {
+      final socket = await WebSocketTransformer.upgrade(request);
+      _sockets.add(socket);
+      socket.add(jsonEncode(['AUTH', 'local-test-challenge']));
+      socket.listen((raw) => _handleSocketMessage(socket, raw));
+      return;
+    }
+    if (request.method == 'POST' && request.uri.path == '/query') {
+      await request.drain<void>();
+      request.response
+        ..statusCode = HttpStatus.ok
+        ..headers.contentType = ContentType.json
+        ..write(jsonEncode(queryEvents));
+      await request.response.close();
+      return;
+    }
+    request.response.statusCode = HttpStatus.notFound;
+    await request.response.close();
+  }
+
+  void _handleSocketMessage(WebSocket socket, Object? raw) {
+    if (raw is! String) return;
+    final decoded = jsonDecode(raw);
+    if (decoded is! List || decoded.isEmpty) return;
+    switch (decoded.first) {
+      case 'AUTH':
+        final event = decoded[1] as Map<String, dynamic>;
+        socket.add(jsonEncode(['OK', event['id'], true, 'authenticated']));
+      case 'REQ':
+        final subId = decoded[1] as String;
+        if (subId.startsWith('h-')) {
+          for (final event in historyEvents) {
+            socket.add(jsonEncode(['EVENT', subId, event]));
+          }
+          socket.add(jsonEncode(['EOSE', subId]));
+        } else {
+          if (!_liveSubscription.isCompleted) {
+            _liveSubscription.complete(subId);
+          }
+          socket.add(jsonEncode(['EOSE', subId]));
+        }
+    }
+  }
+
+  Future<void> sendLive(List<Object?> events) async {
+    final subId = await _liveSubscription.future;
+    final socket = _sockets.single;
+    for (final event in events) {
+      socket.add(jsonEncode(['EVENT', subId, event]));
+    }
+  }
+
+  Future<void> close() async {
+    for (final socket in _sockets) {
+      await socket.close();
+    }
+    await _server.close(force: true);
+  }
+}
+
+class _AuthenticatedAuthNotifier extends AuthNotifier {
+  @override
+  Future<AuthState> build() async =>
+      const AuthState(status: AuthStatus.authenticated);
+}
+
+class _TestRelayConfigNotifier extends RelayConfigNotifier {
+  _TestRelayConfigNotifier(this._origin, this._nsec);
+
+  final String _origin;
+  final String _nsec;
+
+  @override
+  RelayConfig build() => RelayConfig(baseUrl: _origin, nsec: _nsec);
+}
+
+const _channelId = '11111111-1111-4111-8111-111111111111';
+const _filter = NostrFilter(
+  kinds: EventKind.channelEventKinds,
+  tags: {
+    '#h': [_channelId],
+  },
+  limit: 50,
+);
+
+NostrEvent _signedEvent(String content) {
+  final event = nostr.Event.from(
+    kind: EventKind.streamMessageV2,
+    content: content,
+    tags: const [
+      ['h', _channelId],
+    ],
+    secretKey: '1'.padLeft(64, '0'),
+  );
+  return NostrEvent.fromJson(event.toMap());
+}
+
+Map<String, dynamic> _forgedEvent(String content) => {
+  ..._signedEvent(content).toJson(),
+  'sig': List.filled(64, '00').join(),
+};
+
+Future<void> _waitUntil(bool Function() predicate) async {
+  final deadline = DateTime.now().add(const Duration(seconds: 3));
+  while (!predicate()) {
+    if (DateTime.now().isAfter(deadline)) {
+      fail('Timed out waiting for local relay workflow');
+    }
+    await Future<void>.delayed(const Duration(milliseconds: 5));
+  }
+}

mobile/test/shared/relay/relay_session_test.dart

@@ -234,6 +234,137 @@ void main() {
     );
   });
 
+  test('queryRelay accepts a valid signed event', () async {
+    final valid = _signedEvent();
+    final harness = _eventQueryHarness([valid.toJson()]);
+    addTearDown(harness.container.dispose);
+
+    final events = await harness.session.queryRelay(const []);
+
+    expect(events, [valid]);
+  });
+
+  test(
+    'queryRelay rejects a canonical id with a forged zero signature',
+    () async {
+      final forged = {
+        ..._signedEvent().toJson(),
+        'sig': _repeatedHex('00', 64),
+      };
+      final harness = _eventQueryHarness([forged]);
+      addTearDown(harness.container.dispose);
+
+      await expectLater(
+        harness.session.queryRelay(const []),
+        throwsA(isA<FormatException>()),
+      );
+    },
+  );
+
+  final alteredSignedEvents = <String, Map<String, dynamic>>{
+    'altered content': {
+      ..._signedEvent().toJson(),
+      'content': 'attacker content',
+    },
+    'altered id': {..._signedEvent().toJson(), 'id': _repeatedHex('00', 32)},
+    'short id': {..._signedEvent().toJson(), 'id': '00'},
+    'overlong id': {..._signedEvent().toJson(), 'id': _repeatedHex('00', 33)},
+    'nonhex id': {..._signedEvent().toJson(), 'id': _repeatedHex('gg', 32)},
+    'short pubkey': {..._signedEvent().toJson(), 'pubkey': '00'},
+    'overlong pubkey': {
+      ..._signedEvent().toJson(),
+      'pubkey': _repeatedHex('00', 33),
+    },
+    'nonhex pubkey': {
+      ..._signedEvent().toJson(),
+      'pubkey': _repeatedHex('gg', 32),
+    },
+    'short signature': {..._signedEvent().toJson(), 'sig': '00'},
+    'overlong signature': {
+      ..._signedEvent().toJson(),
+      'sig': _repeatedHex('00', 65),
+    },
+    'nonhex signature': {
+      ..._signedEvent().toJson(),
+      'sig': _repeatedHex('gg', 64),
+    },
+  };
+  for (final entry in alteredSignedEvents.entries) {
+    test('queryRelay rejects ${entry.key}', () async {
+      final harness = _eventQueryHarness([entry.value]);
+      addTearDown(harness.container.dispose);
+
+      await expectLater(
+        harness.session.queryRelay(const []),
+        throwsA(isA<FormatException>()),
+      );
+    });
+  }
+
+  test('queryRelay keeps valid events from a mixed invalid batch', () async {
+    final valid = _signedEvent(content: 'valid after forged');
+    final forged = {
+      ..._signedEvent(content: 'forged').toJson(),
+      'sig': _repeatedHex('00', 64),
+    };
+    final harness = _eventQueryHarness([forged, valid.toJson()]);
+    addTearDown(harness.container.dispose);
+
+    final events = await harness.session.queryRelay(const []);
+
+    expect(events, [valid]);
+  });
+
+  test('queryRelay rejects an oversized event before verification', () async {
+    final oversized = {
+      ..._signedEvent().toJson(),
+      'content': List.filled(700000, 'x').join(),
+    };
+    final harness = _eventQueryHarness([oversized]);
+    addTearDown(harness.container.dispose);
+
+    await expectLater(
+      harness.session.queryRelay(const []),
+      throwsA(isA<FormatException>()),
+    );
+  });
+
+  test('queryRelay propagates verification queue exhaustion', () async {
+    final large = {
+      ..._signedEvent().toJson(),
+      'content': List.filled(600000, 'x').join(),
+    };
+    final harness = _eventQueryHarness(List.filled(10, large));
+    addTearDown(harness.container.dispose);
+
+    await expectLater(
+      harness.session.queryRelay(const []),
+      throwsA(isA<StateError>()),
+    );
+  });
+
+  test('queryRelay rejects a response above its byte limit', () async {
+    final harness = _queryHarness(
+      gate: RelayRateLimitGate(),
+      client: http_testing.MockClient(
+        (_) async =>
+            http.Response(List.filled(8 * 1024 * 1024 + 1, 'x').join(), 200),
+      ),
+    );
+    addTearDown(harness.container.dispose);
+
+    await expectLater(
+      harness.session.queryRelay(const []),
+      throwsA(
+        isA<FormatException>().having(
+          (error) => error.message,
+          'message',
+          contains('size limit'),
+        ),
+      ),
+    );
+  });
+
   test('queryRelay rotates the client after a timeout', () async {
     final clients = <_ControlledHttpClient>[];
     final session = RelaySessionNotifier(
@@ -508,6 +639,33 @@ void main() {
     },
   );
 
+  test('websocket history keeps valid events from a mixed batch', () async {
+    final session = RelaySessionNotifier();
+    final history = session.fetchHistory(_channelFilter);
+    final valid = _signedEvent(content: 'valid history');
+    final forged = {
+      ..._signedEvent(content: 'forged history').toJson(),
+      'sig': _repeatedHex('00', 64),
+    };
+
+    session.debugHandleMessage(['EVENT', 'h-1', forged]);
+    session.debugHandleMessage(['EVENT', 'h-1', valid.toJson()]);
+    session.debugHandleMessage(['EOSE', 'h-1']);
+
+    expect(await history, [valid]);
+  });
+
+  test('websocket history rejects an all-invalid response', () async {
+    final session = RelaySessionNotifier();
+    final history = session.fetchHistory(_channelFilter);
+    final forged = {..._signedEvent().toJson(), 'sig': _repeatedHex('00', 64)};
+
+    session.debugHandleMessage(['EVENT', 'h-1', forged]);
+    session.debugHandleMessage(['EOSE', 'h-1']);
+
+    await expectLater(history, throwsA(isA<FormatException>()));
+  });
+
   test('background disconnect rejects in-flight history', () async {
     final session = RelaySessionNotifier();
     final container = ProviderContainer(
@@ -782,16 +940,16 @@ void main() {
     session.debugHandleMessage(['EOSE', 'l-2']);
     final unsubscribeSecond = await secondSubscribe;
 
-    final event = _event();
+    final event = _signedEvent();
     session.debugHandleMessage(['EVENT', 'l-1', event.toJson()]);
     session.debugHandleMessage(['EVENT', 'l-2', event.toJson()]);
-    session.debugFlushEventBuffer();
+    await _waitUntil(() => firstEvents.length == 1 && secondEvents.length == 1);
 
     expect(firstEvents.map((event) => event.id), [event.id]);
     expect(secondEvents.map((event) => event.id), [event.id]);
 
     session.debugHandleMessage(['EVENT', 'l-1', event.toJson()]);
-    session.debugFlushEventBuffer();
+    await Future<void>.delayed(const Duration(milliseconds: 30));
 
     expect(firstEvents.map((event) => event.id), [event.id]);
     expect(secondEvents.map((event) => event.id), [event.id]);
@@ -816,7 +974,7 @@ void main() {
       filter,
       (_) => deliveryPhases.add(queryHasBegun),
     );
-    final replayEvent = _event();
+    final replayEvent = _signedEvent();
     session.debugHandleMessage(['EVENT', 'l-1', replayEvent.toJson()]);
     session.debugHandleMessage(['EOSE', 'l-1']);
 
@@ -830,6 +988,256 @@ void main() {
     unsubscribe();
   });
 
+  test(
+    'live delivery continues with a valid event after an invalid event',
+    () async {
+      final session = RelaySessionNotifier();
+      final delivered = <NostrEvent>[];
+      final subscribe = session.subscribe(_channelFilter, delivered.add);
+      session.debugHandleMessage(['EOSE', 'l-1']);
+      final unsubscribe = await subscribe;
+      final valid = _signedEvent(content: 'valid live');
+      final forged = {
+        ..._signedEvent(content: 'forged live').toJson(),
+        'sig': _repeatedHex('00', 64),
+      };
+
+      session.debugHandleMessage(['EVENT', 'l-1', forged]);
+      session.debugHandleMessage(['EVENT', 'l-1', valid.toJson()]);
+
+      await _waitUntil(() => delivered.isNotEmpty);
+      expect(delivered, [valid]);
+      unsubscribe();
+    },
+  );
+
+  test('EOSE waits for preceding event verification and delivery', () async {
+    final verification = Completer<List<Map<String, dynamic>?>>();
+    var verificationStarted = false;
+    final session = RelaySessionNotifier(
+      verifyEventBatch: (events) {
+        verificationStarted = true;
+        return verification.future;
+      },
+    );
+    final delivered = <NostrEvent>[];
+    var ready = false;
+    final subscribe = session.subscribe(_channelFilter, delivered.add)
+      ..then((_) => ready = true);
+    final valid = _signedEvent();
+
+    session.debugHandleMessage(['EVENT', 'l-1', valid.toJson()]);
+    session.debugHandleMessage(['EOSE', 'l-1']);
+    await _waitUntil(() => verificationStarted);
+    expect(delivered, isEmpty);
+    expect(ready, isFalse);
+
+    verification.complete([valid.toJson()]);
+    final unsubscribe = await subscribe;
+    expect(delivered, [valid]);
+    expect(ready, isTrue);
+    unsubscribe();
+  });
+
+  test('EOSE does not wait for a later event verification', () async {
+    final verifications = <Completer<List<Map<String, dynamic>?>>>[];
+    final session = RelaySessionNotifier(
+      verifyEventBatch: (events) {
+        final verification = Completer<List<Map<String, dynamic>?>>();
+        verifications.add(verification);
+        return verification.future;
+      },
+    );
+    final delivered = <NostrEvent>[];
+    final subscribe = session.subscribe(_channelFilter, delivered.add);
+    final beforeEose = _signedEvent(content: 'before EOSE');
+    final afterEose = _signedEvent(content: 'after EOSE');
+
+    session.debugHandleMessage(['EVENT', 'l-1', beforeEose.toJson()]);
+    session.debugHandleMessage(['EOSE', 'l-1']);
+    await _waitUntil(() => verifications.length == 1);
+    session.debugHandleMessage(['EVENT', 'l-1', afterEose.toJson()]);
+    verifications.first.complete([beforeEose.toJson()]);
+
+    final unsubscribe = await subscribe.timeout(
+      const Duration(milliseconds: 200),
+    );
+    expect(delivered, [beforeEose]);
+    await _waitUntil(() => verifications.length == 2);
+    unsubscribe();
+    verifications.last.complete([afterEose.toJson()]);
+    await Future<void>.delayed(const Duration(milliseconds: 30));
+    expect(delivered, [beforeEose]);
+  });
+
+  test('disconnect immediately retires pending live verification', () async {
+    final verification = Completer<List<Map<String, dynamic>?>>();
+    var verificationStarted = false;
+    final session = RelaySessionNotifier(
+      verifyEventBatch: (events) {
+        verificationStarted = true;
+        return verification.future;
+      },
+    );
+    final container = ProviderContainer(
+      overrides: [
+        authProvider.overrideWith(() => _PendingAuthNotifier()),
+        relaySessionProvider.overrideWith(() => session),
+        relayConfigProvider.overrideWith(
+          () => _FakeRelayConfigNotifier(
+            baseUrl: 'https://relay.example',
+            nsec: nostr.Keys.generate().nsec,
+          ),
+        ),
+      ],
+    );
+    addTearDown(container.dispose);
+    container.read(relaySessionProvider);
+    final delivered = <NostrEvent>[];
+    final subscribe = session.subscribe(_channelFilter, delivered.add);
+    session.debugHandleMessage(['EOSE', 'l-1']);
+    final unsubscribe = await subscribe;
+    final valid = _signedEvent();
+
+    session.debugHandleMessage(['EVENT', 'l-1', valid.toJson()]);
+    await _waitUntil(() => verificationStarted);
+    session.debugHandleDisconnected();
+    verification.complete([valid.toJson()]);
+    await Future<void>.delayed(const Duration(milliseconds: 30));
+    session.debugFlushEventBuffer();
+
+    expect(delivered, isEmpty);
+    unsubscribe();
+  });
+
+  test('unsubscribe immediately retires pending live verification', () async {
+    final verification = Completer<List<Map<String, dynamic>?>>();
+    var verificationStarted = false;
+    final session = RelaySessionNotifier(
+      verifyEventBatch: (events) {
+        verificationStarted = true;
+        return verification.future;
+      },
+    );
+    final delivered = <NostrEvent>[];
+    final subscribe = session.subscribe(_channelFilter, delivered.add);
+    session.debugHandleMessage(['EOSE', 'l-1']);
+    final unsubscribe = await subscribe;
+    final valid = _signedEvent();
+
+    session.debugHandleMessage(['EVENT', 'l-1', valid.toJson()]);
+    await _waitUntil(() => verificationStarted);
+    unsubscribe();
+    verification.complete([valid.toJson()]);
+    await Future<void>.delayed(const Duration(milliseconds: 30));
+    session.debugFlushEventBuffer();
+
+    expect(delivered, isEmpty);
+  });
+
+  test('terminal CLOSED immediately retires pending verification', () async {
+    final verification = Completer<List<Map<String, dynamic>?>>();
+    var verificationStarted = false;
+    final session = RelaySessionNotifier(
+      verifyEventBatch: (events) {
+        verificationStarted = true;
+        return verification.future;
+      },
+    );
+    final delivered = <NostrEvent>[];
+    final subscribe = session.subscribe(_channelFilter, delivered.add);
+    session.debugHandleMessage(['EOSE', 'l-1']);
+    await subscribe;
+    final valid = _signedEvent();
+
+    session.debugHandleMessage(['EVENT', 'l-1', valid.toJson()]);
+    await _waitUntil(() => verificationStarted);
+    session.debugHandleMessage(['CLOSED', 'l-1', 'restricted: removed']);
+    verification.complete([valid.toJson()]);
+    await Future<void>.delayed(const Duration(milliseconds: 30));
+    session.debugFlushEventBuffer();
+
+    expect(delivered, isEmpty);
+  });
+
+  test('local OK acknowledgment bypasses pending verification', () async {
+    final verification = Completer<List<Map<String, dynamic>?>>();
+    var verificationStarted = false;
+    final session = RelaySessionNotifier(
+      verifyEventBatch: (events) {
+        verificationStarted = true;
+        return verification.future;
+      },
+    );
+    final subscribe = session.subscribe(_channelFilter, (_) {});
+    session.debugHandleMessage(['EOSE', 'l-1']);
+    final unsubscribe = await subscribe;
+    session.debugHandleMessage(['EVENT', 'l-1', _signedEvent().toJson()]);
+    await _waitUntil(() => verificationStarted);
+    session.debugAttachSocketForTest(_RecordingRelaySocket());
+
+    final publish = session.publish(_event(id: 'local-event'));
+    session.debugHandleMessage(['OK', 'local-event', true, 'accepted']);
+
+    expect(
+      (await publish.timeout(const Duration(milliseconds: 100))).id,
+      'local-event',
+    );
+    unsubscribe();
+    verification.complete([_signedEvent().toJson()]);
+  });
+
+  test('session teardown fences a pending verification result', () async {
+    final verification = Completer<List<Map<String, dynamic>?>>();
+    var verificationStarted = false;
+    final session = RelaySessionNotifier(
+      verifyEventBatch: (events) {
+        verificationStarted = true;
+        return verification.future;
+      },
+    );
+    final delivered = <NostrEvent>[];
+    final subscribe = session.subscribe(_channelFilter, delivered.add);
+    session.debugHandleMessage(['EOSE', 'l-1']);
+    await subscribe;
+    final valid = _signedEvent();
+
+    session.debugHandleMessage(['EVENT', 'l-1', valid.toJson()]);
+    await _waitUntil(() => verificationStarted);
+    session.debugDispose();
+    verification.complete([valid.toJson()]);
+    await Future<void>.delayed(const Duration(milliseconds: 30));
+    session.debugFlushEventBuffer();
+
+    expect(delivered, isEmpty);
+  });
+
+  test('connection replacement fences a pending verification result', () async {
+    final verification = Completer<List<Map<String, dynamic>?>>();
+    var verificationStarted = false;
+    final session = RelaySessionNotifier(
+      verifyEventBatch: (events) {
+        verificationStarted = true;
+        return verification.future;
+      },
+    );
+    final delivered = <NostrEvent>[];
+    final subscribe = session.subscribe(_channelFilter, delivered.add);
+    session.debugHandleMessage(['EOSE', 'l-1']);
+    final unsubscribe = await subscribe;
+    final valid = _signedEvent();
+
+    session.debugHandleMessage(['EVENT', 'l-1', valid.toJson()]);
+    await _waitUntil(() => verificationStarted);
+    session.debugSupersedeConnection();
+    verification.complete([valid.toJson()]);
+    await Future<void>.delayed(const Duration(milliseconds: 30));
+    session.debugFlushEventBuffer();
+
+    expect(delivered, isEmpty);
+    unsubscribe();
+  });
+
   test('terminal CLOSED fails a live subscribe before ready', () async {
     final session = RelaySessionNotifier();
     const filter = NostrFilter(kinds: [EventKind.agentObserverFrame], limit: 0);
@@ -919,7 +1327,7 @@ void main() {
     session.debugHandleMessage([
       'EVENT',
       'l-1',
-      _event(createdAt: 30).toJson(),
+      _signedEvent(createdAt: 30).toJson(),
     ]);
     expect(statuses, [
       RelaySubscriptionStatus.ready,
@@ -927,6 +1335,7 @@ void main() {
     ]);
 
     session.debugHandleMessage(['EOSE', 'l-1']);
+    await _waitUntil(() => statuses.length == 3);
     expect(statuses, [
       RelaySubscriptionStatus.ready,
       RelaySubscriptionStatus.retrying,
@@ -1004,7 +1413,8 @@ void main() {
       },
     );
     session.debugAttachSocketForTest(socket);
-    final subscribe = session.subscribe(_channelFilter, (_) {});
+    final delivered = <NostrEvent>[];
+    final subscribe = session.subscribe(_channelFilter, delivered.add);
     session.debugHandleMessage(['EOSE', 'l-1']);
     final unsubscribe = await subscribe;
 
@@ -1017,8 +1427,9 @@ void main() {
     session.debugHandleMessage([
       'EVENT',
       'l-1',
-      _event(createdAt: 30).toJson(),
+      _signedEvent(createdAt: 30).toJson(),
     ]);
+    await _waitUntil(() => delivered.isNotEmpty);
     session.debugHandleMessage(['CLOSED', 'l-1', 'error: transient']);
     expect(timers.last.duration, const Duration(seconds: 1));
     unsubscribe();
@@ -1772,6 +2183,13 @@ _QueryHarness _queryHarness({
   return _QueryHarness(container: container, session: session);
 }
 
+_QueryHarness _eventQueryHarness(List<Object?> events) => _queryHarness(
+  gate: RelayRateLimitGate(),
+  client: http_testing.MockClient(
+    (_) async => http.Response(jsonEncode(events), 200),
+  ),
+);
+
 class _FakeAuthNotifier extends AuthNotifier {
   int signOutCount = 0;
 
@@ -1846,6 +2264,31 @@ NostrEvent _event({int createdAt = 20, String id = 'event-1'}) {
   );
 }
 
+NostrEvent _signedEvent({int createdAt = 20, String content = 'hello'}) {
+  final event = nostr.Event.from(
+    kind: EventKind.streamMessageV2,
+    content: content,
+    tags: const [
+      ['h', _channelId],
+    ],
+    createdAt: createdAt,
+    secretKey: '1'.padLeft(64, '0'),
+  );
+  return NostrEvent.fromJson(event.toMap());
+}
+
+String _repeatedHex(String pair, int bytes) => List.filled(bytes, pair).join();
+
+Future<void> _waitUntil(bool Function() predicate) async {
+  final deadline = DateTime.now().add(const Duration(seconds: 2));
+  while (!predicate()) {
+    if (DateTime.now().isAfter(deadline)) {
+      fail('Timed out waiting for asynchronous relay work');
+    }
+    await Future<void>.delayed(const Duration(milliseconds: 1));
+  }
+}
+
 const _visibleChannelId = '99999999-9999-4999-8999-999999999999';
 const _channelFilter = NostrFilter(
   kinds: EventKind.channelEventKinds,