From 06a016e1cb69e9fbdcda9826a762c08ea800cba9 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Wed, 9 Sep 2026 22:29:27 -0400 Subject: [PATCH] fix(mobile): preserve SEND scope and authority at owner 7536 Signed-off-by: Logan Johnson --- .../compose_bar/compose_bar_widget.dart | 108 ++++---- .../channels/compose_bar/helpers.dart | 22 +- .../selected_mention_preparation.dart | 72 ++++- mobile/lib/shared/mentions/agent_policy.dart | 50 ++-- .../selected_mention_authorization.dart | 61 ++++- .../shared/profile/user_cache_provider.dart | 4 - mobile/lib/shared/relay/relay_session.dart | 17 +- .../mentions/selected_observation_test.dart | 250 ++++++++++++++++++ 8 files changed, 480 insertions(+), 104 deletions(-) create mode 100644 mobile/test/shared/mentions/selected_observation_test.dart diff --git a/mobile/lib/features/channels/compose_bar/compose_bar_widget.dart b/mobile/lib/features/channels/compose_bar/compose_bar_widget.dart index a1213e895d1..28c5869f85b 100644 --- a/mobile/lib/features/channels/compose_bar/compose_bar_widget.dart +++ b/mobile/lib/features/channels/compose_bar/compose_bar_widget.dart @@ -485,12 +485,7 @@ class ComposeBar extends HookConsumerWidget { var authorizationRevision = submittedDraftRevision; final visit = authorizationVisit.value; final config = ref.read(relayConfigProvider); - // Equivalent refreshes retain scope; destination/credentials do not. - bool isConfigScopeCurrent() { - final current = ref.read(relayConfigProvider); - return current.baseUrl == config.baseUrl && - current.nsec == config.nsec; - } + bool isConfigScopeCurrent() => _isComposeConfigCurrent(ref, config); final readSelected = ref.read( selectedMentionAuthorizationReaderProvider, @@ -498,29 +493,20 @@ class ComposeBar extends HookConsumerWidget { final session = ref.read(relaySessionProvider.notifier); final observedProfiles = {}; final observedKeys = {}; - bool profilesCurrent() => observedProfiles.entries.every((entry) { - final order = ref - .read(userCacheProvider.notifier) - .profileEventOrder(entry.key); - final event = entry.value; - return order == null || - order.createdAt < event.createdAt || - (order.createdAt == event.createdAt && - order.eventId.compareTo(event.id) >= 0); - }); + final evidenceChecks = {}; + bool profilesCurrent() => evidenceChecks.every((check) => check()); bool ownsSource() => context.mounted && visit == authorizationVisit.value && isConfigScopeCurrent(); - bool isAuthorizationCurrent() => + bool isAuthorizationScopeCurrent() => ownsSource() && identical(session, ref.read(relaySessionProvider.notifier)) && currentPubkey == ref.read(currentPubkeyProvider) && - profilesCurrent() && submittedUploadGeneration == uploadGeneration.value && authorizationRevision == draftRevision.value && isConfigScopeCurrent(); - void ensureAuthorizationCurrent() { + void ensureAuthorizationScopeCurrent() { if (!context.mounted) throw const _ComposeAuthorizationCancelled(); if (!isConfigScopeCurrent()) { throw const _ComposeCommunityChanged(); @@ -531,12 +517,26 @@ class ComposeBar extends HookConsumerWidget { throw const _ComposeAuthorizationCancelled(); } if (!identical(session, ref.read(relaySessionProvider.notifier)) || - currentPubkey != ref.read(currentPubkeyProvider) || - !profilesCurrent()) { + currentPubkey != ref.read(currentPubkeyProvider)) { + throw Exception('Mention evidence changed; retry the draft'); + } + } + + void ensureAuthorizationCurrent() { + ensureAuthorizationScopeCurrent(); + if (!profilesCurrent()) { throw Exception('Mention evidence changed; retry the draft'); } } + Future guardedDelivery( + String content, + List keys, { + List> mediaTags = const [], + }) => withRelayPublicationGuard( + ensureAuthorizationCurrent, + () => onSend(content, keys, mediaTags: mediaTags), + ); checkPreparationCurrent = ensureAuthorizationCurrent; // Resolved before any await: see @@ -566,10 +566,12 @@ class ComposeBar extends HookConsumerWidget { priorAgentKeys: priorAgentKeys, observedKeys: observedKeys, observedProfiles: observedProfiles, + evidenceChecks: evidenceChecks, currentPubkey: currentPubkey, channelId: channelId, ensureAuthorizationCurrent: ensureAuthorizationCurrent, - isAuthorizationCurrent: isAuthorizationCurrent, + isAuthorizationCurrent: isAuthorizationScopeCurrent, + ensureScopeCurrent: ensureAuthorizationScopeCurrent, prepare: prepare, ); @@ -610,44 +612,21 @@ class ComposeBar extends HookConsumerWidget { ); final channelActions = ref.read(channelActionsProvider); - // Agent failures stop publication; the original draft keeps its keys. - Future addMentionedNonMembers() async { - final keys = outgoing.pubkeys.toSet(); - Future authorizeWrite(String key, String role) async { - final evidence = await authorize(keys, prepare: true); - final fresh = evidence[key]!; - if (fresh.invitationRole != role) { - throw Exception( - 'Mention classification changed; retry invitation consent', - ); - } - return !fresh.isMember; - } - - await authorize(keys, prepare: true); - ensureAuthorizationCurrent(); - invitationStarted.value = true; - await outgoing.addNonMembers( - channelActions, - scan: scan, - messenger: messenger, - ensureCurrent: ensureAuthorizationCurrent, - authorizeWrite: authorizeWrite, - ); - if (!outgoing.pubkeys.toSet().containsAll(keys)) { - throw Exception( - 'Mention invitation failed. Draft kept; retry or remove the mention.', - ); - } - await authorize(keys); - if (queuedAttachments.isEmpty || - (selectedKeys.isNotEmpty || - scan.humans.isNotEmpty || - scan.agentPubkeys.isNotEmpty)) { - ensureAuthorizationCurrent(); - } - } - + Future addMentionedNonMembers() => _prepareMentionInvitations( + outgoing: outgoing, + scan: scan, + channelActions: channelActions, + messenger: messenger, + authorize: authorize, + ensureCurrent: ensureAuthorizationCurrent, + ensureScopeCurrent: ensureAuthorizationScopeCurrent, + onStarted: () => invitationStarted.value = true, + fenceAfterPreparation: + queuedAttachments.isEmpty || + selectedKeys.isNotEmpty || + scan.humans.isNotEmpty || + scan.agentPubkeys.isNotEmpty, + ); if (queuedAttachments.isEmpty) { if (!context.mounted) return; await _sendTextOnlyDraft( @@ -658,7 +637,10 @@ class ComposeBar extends HookConsumerWidget { submittedDraftRevision: submittedDraftRevision, ownsSource: ownsSource, focusNode: focusNode, - clearComposer: clearComposer, + clearComposer: () { + clearComposer(); + authorizationRevision = draftRevision.value; + }, addMentionedNonMembers: addMentionedNonMembers, payload: _ComposeDraftPayload.fromDraft( text: text, @@ -666,7 +648,7 @@ class ComposeBar extends HookConsumerWidget { customEmoji: customEmoji, ), outgoing: outgoing, - onSend: onSend, + onSend: guardedDelivery, messenger: messenger, ); return; @@ -691,7 +673,7 @@ class ComposeBar extends HookConsumerWidget { final cancellation = UploadCancellationToken(); final uploadService = ref.read(mediaUploadServiceProvider); activeUploadCancellation.value = cancellation; - final delivery = onSend; + final delivery = guardedDelivery; unawaited(() async { var retainedForRetry = false; var delivered = false; diff --git a/mobile/lib/features/channels/compose_bar/helpers.dart b/mobile/lib/features/channels/compose_bar/helpers.dart index 0bd8e7395d1..17a79caa506 100644 --- a/mobile/lib/features/channels/compose_bar/helpers.dart +++ b/mobile/lib/features/channels/compose_bar/helpers.dart @@ -479,6 +479,7 @@ Future<_NonMemberAddOutcome> _addMentionedNonMembers( required List humanPubkeys, required bool canAddMembers, required VoidCallback ensureCurrent, + required VoidCallback ensureScopeCurrent, required VoidCallback onAccepted, required Future Function(String, String) authorizeWrite, }) async { @@ -500,23 +501,26 @@ Future<_NonMemberAddOutcome> _addMentionedNonMembers( final notAdded = []; final errors = []; for (final (pubkeys, role) in pending) { - ensureCurrent(); + ensureScopeCurrent(); if (!await authorizeWrite(pubkeys.single, role)) continue; ensureCurrent(); try { - await channelActions.addMembers( - channelId: channelId, - pubkeys: pubkeys, - role: role, - onAccepted: (_) => onAccepted(), + await withRelayPublicationGuard( + ensureCurrent, + () => channelActions.addMembers( + channelId: channelId, + pubkeys: pubkeys, + role: role, + onAccepted: (_) => onAccepted(), + ), ); - ensureCurrent(); + ensureScopeCurrent(); } on _ComposeAuthorizationCancelled { rethrow; } on _ComposeCommunityChanged { rethrow; } on StateError { - ensureCurrent(); + ensureScopeCurrent(); rethrow; } catch (error) { notAdded.addAll( @@ -669,6 +673,7 @@ class _OutgoingMentions { required _NonMemberMentionScan scan, required ScaffoldMessengerState? messenger, required VoidCallback ensureCurrent, + required VoidCallback ensureScopeCurrent, required Future Function(String, String) authorizeWrite, }) async { final outcome = await _addMentionedNonMembers( @@ -680,6 +685,7 @@ class _OutgoingMentions { .toList(), canAddMembers: scan.canAddMembers, ensureCurrent: ensureCurrent, + ensureScopeCurrent: ensureScopeCurrent, onAccepted: () => acceptedInvitations++, authorizeWrite: authorizeWrite, ); diff --git a/mobile/lib/features/channels/compose_bar/selected_mention_preparation.dart b/mobile/lib/features/channels/compose_bar/selected_mention_preparation.dart index f5a92ba3744..4eca9d35d6f 100644 --- a/mobile/lib/features/channels/compose_bar/selected_mention_preparation.dart +++ b/mobile/lib/features/channels/compose_bar/selected_mention_preparation.dart @@ -8,14 +8,19 @@ Future> _authorizeSelectedMentions( required Set priorAgentKeys, required Set observedKeys, required Map observedProfiles, + required Set evidenceChecks, required String? currentPubkey, required String channelId, required VoidCallback ensureAuthorizationCurrent, + required VoidCallback ensureScopeCurrent, required bool Function() isAuthorizationCurrent, bool prepare = false, }) async { - ensureAuthorizationCurrent(); - if (keys.isEmpty) return const {}; + ensureScopeCurrent(); + if (keys.isEmpty) { + evidenceChecks.clear(); + return const {}; + } try { final evidence = await readSelected( keys, @@ -34,11 +39,15 @@ Future> _authorizeSelectedMentions( observedProfiles.addAll(profiles); }, ); - ensureAuthorizationCurrent(); + ensureScopeCurrent(); if (evidence.length != keys.length || !evidence.keys.toSet().containsAll(keys)) { throw Exception('Incomplete selected mention evidence'); } + evidenceChecks + ..clear() + ..addAll(evidence.values.map((value) => value.isCurrent)); + ensureAuthorizationCurrent(); final agents = { for (final key in keys) if (evidence[key]!.requiresAgentAuthorization) key, @@ -55,7 +64,62 @@ Future> _authorizeSelectedMentions( ensureAuthorizationCurrent(); return evidence; } catch (_) { - ensureAuthorizationCurrent(); + ensureScopeCurrent(); rethrow; } } + +// Failures preserve the exact draft audience; consent never licenses a changed +// role. Kept beside the reader owner so both boundaries use one evaluator. +Future _prepareMentionInvitations({ + required _OutgoingMentions outgoing, + required _NonMemberMentionScan scan, + required ChannelActions channelActions, + required ScaffoldMessengerState? messenger, + required Future> Function( + Set keys, { + bool prepare, + }) + authorize, + required VoidCallback ensureCurrent, + required VoidCallback ensureScopeCurrent, + required VoidCallback onStarted, + required bool fenceAfterPreparation, +}) async { + final keys = outgoing.pubkeys.toSet(); + Future authorizeWrite(String key, String role) async { + final evidence = await authorize(keys, prepare: true); + final fresh = evidence[key]!; + if (fresh.invitationRole != role) { + throw Exception( + 'Mention classification changed; retry invitation consent', + ); + } + return !fresh.isMember; + } + + await authorize(keys, prepare: true); + ensureCurrent(); + onStarted(); + await outgoing.addNonMembers( + channelActions, + scan: scan, + messenger: messenger, + ensureCurrent: ensureCurrent, + ensureScopeCurrent: ensureScopeCurrent, + authorizeWrite: authorizeWrite, + ); + if (!outgoing.pubkeys.toSet().containsAll(keys)) { + throw Exception( + 'Mention invitation failed. Draft kept; retry or remove the mention.', + ); + } + await authorize(keys); + if (fenceAfterPreparation) ensureCurrent(); +} + +// Equivalent refreshes retain scope; destination/credentials do not. +bool _isComposeConfigCurrent(WidgetRef ref, RelayConfig config) { + final current = ref.read(relayConfigProvider); + return current.baseUrl == config.baseUrl && current.nsec == config.nsec; +} diff --git a/mobile/lib/shared/mentions/agent_policy.dart b/mobile/lib/shared/mentions/agent_policy.dart index 0abc2e076bb..80798f22794 100644 --- a/mobile/lib/shared/mentions/agent_policy.dart +++ b/mobile/lib/shared/mentions/agent_policy.dart @@ -11,13 +11,16 @@ Future> _queryAgentFilters( RelaySessionNotifier session, List filters, { void Function()? checkCurrent, + void Function(List, List?)? onQueryEvidence, }) async { final events = []; + onQueryEvidence?.call(filters, null); for (var start = 0; start < filters.length; start += 10) { checkCurrent?.call(); - events.addAll( - await session.queryRelay(filters.skip(start).take(10).toList()), - ); + final batch = filters.skip(start).take(10).toList(); + final result = await session.queryRelay(batch); + onQueryEvidence?.call(batch, result); + events.addAll(result); } return events; } @@ -32,6 +35,7 @@ Future> resolveAgentPolicies( Set? requestedKeys, void Function()? checkCurrent, void Function(Map profiles)? onProfileEvidence, + void Function(List, List?)? onQueryEvidence, }) async { final latest = {}; for (final event in runtimeEvents.where((event) => event.kind == 10100)) { @@ -42,10 +46,15 @@ Future> resolveAgentPolicies( } final keys = requestedKeys ?? latest.keys.toSet(); final profiles = latestProfileEvents( - await _queryAgentFilters(session, [ - for (final key in keys) - NostrFilter(kinds: const [0], authors: [key], limit: 1), - ], checkCurrent: checkCurrent), + await _queryAgentFilters( + session, + [ + for (final key in keys) + NostrFilter(kinds: const [0], authors: [key], limit: 1), + ], + checkCurrent: checkCurrent, + onQueryEvidence: onQueryEvidence, + ), ); final owners = {}; for (final profile in profiles.values) { @@ -55,17 +64,22 @@ Future> resolveAgentPolicies( } } checkCurrent?.call(); - final policies = await _queryAgentFilters(session, [ - for (final owner in owners.entries) - NostrFilter( - kinds: const [30177], - authors: [owner.value], - tags: { - '#d': [owner.key], - }, - limit: 1, - ), - ], checkCurrent: checkCurrent); + final policies = await _queryAgentFilters( + session, + [ + for (final owner in owners.entries) + NostrFilter( + kinds: const [30177], + authors: [owner.value], + tags: { + '#d': [owner.key], + }, + limit: 1, + ), + ], + checkCurrent: checkCurrent, + onQueryEvidence: onQueryEvidence, + ); checkCurrent?.call(); onProfileEvidence?.call(Map.unmodifiable(profiles)); return mergeAgentPolicies(latest.values, policies, owners); diff --git a/mobile/lib/shared/mentions/selected_mention_authorization.dart b/mobile/lib/shared/mentions/selected_mention_authorization.dart index 152426234d6..a45a55432ee 100644 --- a/mobile/lib/shared/mentions/selected_mention_authorization.dart +++ b/mobile/lib/shared/mentions/selected_mention_authorization.dart @@ -1,5 +1,7 @@ part of 'agent_identity_provider.dart'; +bool _unfencedSelectedEvidence() => true; + /// Evidence requirements, not a claim that an ordinary identity is human. enum SelectedMentionKind { /// No fresh agent evidence for this key. Absence of agent evidence, not @@ -36,8 +38,16 @@ class SelectedMentionAuthorization { /// owner policy produces a deny-all entry, never runtime fallback. final AgentDirectoryEntry? agent; + /// Local observed evidence capability, retained through actual enqueue. + final bool Function() isCurrent; + /// Construct evidence, not a publication or role-write capability. - const SelectedMentionAuthorization(this.kind, this.isMember, this.agent); + const SelectedMentionAuthorization( + this.kind, + this.isMember, + this.agent, { + this.isCurrent = _unfencedSelectedEvidence, + }); /// True when this key's evidence must be routed through the agent /// authorization evaluator rather than the ordinary flow: agent or @@ -73,8 +83,36 @@ readSelectedMentionAuthorization( required bool Function() isCurrent, void Function(Map)? onProfileEvidence, }) async { + final fences = [session.retainPublicationEvidence()]; + bool evidenceCurrent() => fences.every((check) => check()); void check() { - if (!isCurrent()) throw StateError('Mention authorization scope changed'); + if (!isCurrent() || !evidenceCurrent()) { + throw StateError('Mention authorization scope changed'); + } + } + + void observeQuery(List filters, List? events) { + for (final filter in filters) { + for (final kind in filter.kinds) { + for (final author in filter.authors!) { + fences.add( + events == null + ? session.evidenceClock.retain( + kind, + author, + filter.tags['#d']?.single, + ) + : session.evidenceClock.snapshot( + kind, + author, + filter.tags['#d']?.single, + events, + ), + ); + } + } + } + check(); } check(); @@ -88,6 +126,7 @@ readSelectedMentionAuthorization( if (requestedKeys.isEmpty) return const {}; final authority = await session.fetchRelaySelf(); check(); + fences.add(session.evidenceClock.retain(39002, authority, channelId)); final membership = await _membershipPages( session, authority, @@ -95,6 +134,9 @@ readSelectedMentionAuthorization( channelId, check, ); + fences.add( + session.evidenceClock.snapshot(39002, authority, channelId, membership), + ); check(); NostrEvent? roster; for (final event in membership) { @@ -120,10 +162,15 @@ readSelectedMentionAuthorization( } members[tag[1]] = tag.length >= 4 ? tag[3] : 'member'; } - final runtime = await _queryAgentFilters(session, [ - for (final key in requestedKeys) - NostrFilter(kinds: const [10100], authors: [key], limit: 1), - ], checkCurrent: check); + final runtime = await _queryAgentFilters( + session, + [ + for (final key in requestedKeys) + NostrFilter(kinds: const [10100], authors: [key], limit: 1), + ], + checkCurrent: check, + onQueryEvidence: observeQuery, + ); check(); Map profiles = const {}; final policies = await resolveAgentPolicies( @@ -132,6 +179,7 @@ readSelectedMentionAuthorization( requestedKeys: requestedKeys, checkCurrent: check, onProfileEvidence: (value) => profiles = value, + onQueryEvidence: observeQuery, ); onProfileEvidence?.call(profiles); check(); @@ -193,6 +241,7 @@ readSelectedMentionAuthorization( kind, members.containsKey(key), agent, + isCurrent: evidenceCurrent, ); } check(); diff --git a/mobile/lib/shared/profile/user_cache_provider.dart b/mobile/lib/shared/profile/user_cache_provider.dart index e54083aa269..91131c888d0 100644 --- a/mobile/lib/shared/profile/user_cache_provider.dart +++ b/mobile/lib/shared/profile/user_cache_provider.dart @@ -31,10 +31,6 @@ class UserCacheNotifier extends Notifier> { return {}; } - /// Latest observed profile revision; read-only fencing for publication. - ({int createdAt, String eventId})? profileEventOrder(String key) => - _profileEventOrders[key]; - /// Request a profile for [pubkey]. Returns immediately from cache if /// available, otherwise schedules a batch fetch. UserProfile? get(String pubkey) { diff --git a/mobile/lib/shared/relay/relay_session.dart b/mobile/lib/shared/relay/relay_session.dart index b0698416932..49a6fe08725 100644 --- a/mobile/lib/shared/relay/relay_session.dart +++ b/mobile/lib/shared/relay/relay_session.dart @@ -13,6 +13,7 @@ import 'package:hooks_riverpod/hooks_riverpod.dart'; import '../auth/auth.dart'; import 'nostr_models.dart'; import 'relay_client.dart'; +import 'relay_evidence_clock.dart'; import 'relay_closed_policy.dart'; import 'relay_http_query_client.dart'; import 'relay_provider.dart'; @@ -110,6 +111,9 @@ class RelaySessionNotifier extends Notifier { _retryTimerFactory = retryTimerFactory, _replayDelay = replayDelay; + /// Immediate evidence ordering shared by all reads on this session. + final evidenceClock = RelayEvidenceClock(); + final RelayHttpQueryClient _httpQueryClient; final RelaySocketFactory _socketFactory; final DateTime Function() _now; @@ -156,6 +160,7 @@ class RelaySessionNotifier extends Notifier { // Reset disposed flag — build() may re-run on the same Notifier instance // after a provider dependency changes (e.g. auth completing). _disposed = false; + evidenceClock.clear(); ref.onDispose(_dispose); @@ -194,6 +199,7 @@ class RelaySessionNotifier extends Notifier { Duration timeout = const Duration(seconds: 8), }) async { final config = ref.read(relayConfigProvider); + final generation = _connectionGeneration; final url = Uri.parse(config.baseUrl).resolve('/query').toString(); final bodyBytes = utf8.encode( jsonEncode(filters.map((filter) => filter.toJson()).toList()), @@ -223,13 +229,19 @@ class RelaySessionNotifier extends Notifier { throw const FormatException('relay returned malformed query response'); } try { - return [ + final events = [ for (final eventJson in decoded) if (eventJson is Map) NostrEvent.fromJson(eventJson) else throw const FormatException('relay returned malformed query event'), ]; + if (_isActiveConnection(generation)) { + for (final event in events) { + evidenceClock.observe(event); + } + } + return events; } catch (error) { if (error is FormatException) rethrow; throw FormatException('relay returned malformed query event: $error'); @@ -409,6 +421,7 @@ class RelaySessionNotifier extends Notifier { void debugSupersedeConnection() => _supersedeConnection(); int _supersedeConnection() { + evidenceClock.clear(); return ++_connectionGeneration; } @@ -700,6 +713,7 @@ class RelaySessionNotifier extends Notifier { final subId = data[1] as String; final eventJson = data[2] as Map; final event = NostrEvent.fromJson(eventJson); + evidenceClock.observe(event); // History subscriptions accumulate immediately. final historySub = _historySubscriptions[subId]; @@ -1016,6 +1030,7 @@ class RelaySessionNotifier extends Notifier { } void _dispose() { + evidenceClock.clear(); _disposed = true; _beforePauseCallbacks.clear(); _connectionGeneration++; diff --git a/mobile/test/shared/mentions/selected_observation_test.dart b/mobile/test/shared/mentions/selected_observation_test.dart new file mode 100644 index 00000000000..64e48caca3e --- /dev/null +++ b/mobile/test/shared/mentions/selected_observation_test.dart @@ -0,0 +1,250 @@ +import 'dart:async'; +import 'dart:convert'; + +import 'package:buzz/shared/auth/auth.dart'; +import 'package:buzz/shared/mentions/agent_identity_provider.dart'; +import 'package:buzz/shared/relay/relay.dart'; +import 'package:buzz/shared/relay/relay_evidence_clock.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:hooks_riverpod/hooks_riverpod.dart'; +import 'package:http/http.dart' as http; +import 'package:http/testing.dart'; +import 'package:nostr/nostr.dart' as nostr; + +import '../crypto/nip_oa_test.dart' show authTag, profile; +import 'agent_policy_test.dart' show signed; + +void main() { + test( + 'coordinate eviction retires in-flight proof without global cancellation', + () { + final clock = RelayEvidenceClock(); + final owner = nostr.Keys.generate(); + final retained = clock.retain(0, owner.public, null); + for (var i = 0; i < 8192; i++) { + clock.retain(30177, owner.public, '$i'); + } + expect(retained(), isFalse); + final unrelated = clock.snapshot(30177, owner.public, '8191', []); + expect(unrelated(), isTrue); + }, + ); + + for (final scenario in [ + 'policy', + 'runtime', + 'missing profile', + 'roster', + 'unrelated', + 'retired session', + 'replacement socket', + 'queued policy', + 'queued missing profile', + ]) { + final mode = scenario.replaceFirst('queued ', ''); + final queued = scenario.startsWith('queued '); + test( + 'production reader rejects stale completion after observed $scenario', + () async { + final viewer = nostr.Keys.generate(); + final agent = nostr.Keys.generate(); + final relay = nostr.Keys.generate(); + final other = nostr.Keys.generate(); + NostrEvent policy(String mode, int time) => signed( + viewer, + 30177, + {'name': 'Agent', 'parallelism': 1, 'respond_to': mode}, + time: time, + tags: [ + ['d', agent.public], + ], + ); + NostrEvent roster(int time) => signed( + relay, + 39002, + '', + time: time, + tags: [ + ['d', 'room'], + ['p', viewer.public], + [ + 'p', + agent.public, + '', + mode == 'missing profile' ? 'member' : 'bot', + ], + ], + ); + final events = [ + roster(100), + if (mode != 'missing profile') + profile(agent, [authTag(viewer, agent.public)]), + if (mode != 'missing profile') + signed(agent, 10100, {'respond_to': 'anyone'}, time: 100), + policy('anyone', 100), + ]; + final oldQuery = Completer(); + final release = Completer(); + final blockedKind = switch (mode) { + 'runtime' => 10100, + 'missing profile' => 0, + 'roster' => 39002, + _ => 30177, + }; + final client = MockClient((request) async { + if (request.method == 'GET') { + return http.Response(jsonEncode({'self': relay.public}), 200); + } + final filters = (jsonDecode(request.body) as List) + .cast>(); + final results = events + .where( + (event) => filters.any( + (filter) => + (filter['kinds'] as List).contains(event.kind) && + (filter['authors'] as List).contains(event.pubkey) && + (filter['#d'] == null || + (filter['#d'] as List).contains( + event.getTagValue('d'), + )), + ), + ) + .toList(); + if ((filters.first['kinds'] as List).contains(blockedKind)) { + oldQuery.complete(); + await release.future; + } + return http.Response( + jsonEncode(results.map((e) => e.toJson()).toList()), + 200, + ); + }); + final gate = RelayRateLimitGate(); + final session = RelaySessionNotifier( + httpClient: client, + rateLimitGate: gate, + ); + final container = ProviderContainer( + overrides: [ + authProvider.overrideWith(_Auth.new), + relayConfigProvider.overrideWith(() => _Config(viewer.nsec)), + relaySessionProvider.overrideWith(() => session), + ], + ); + addTearDown(container.dispose); + await container.read(authProvider.future); + container.read(relaySessionProvider); + final socket = _Socket(); + session.debugAttachSocketForTest(socket); + final subscription = session.subscribe( + const NostrFilter(kinds: [0, 10100, 30177, 39002]), + (_) {}, + ); + final subId = socket.messages.single[1] as String; + session.debugHandleMessage(['EOSE', subId]); + final unsubscribe = await subscription; + addTearDown(unsubscribe); + final pending = readSelectedMentionAuthorization( + session, + {agent.public}, + viewer: viewer.public, + channelId: 'room', + isCurrent: () => true, + ); + await oldQuery.future; + if (mode == 'retired session' || mode == 'replacement socket') { + final retained = session.evidenceClock.retain(0, agent.public, null); + if (mode == 'retired session') { + container.invalidate(relaySessionProvider); + container.read(relaySessionProvider); + } else { + session.debugSupersedeConnection(); + } + expect(retained(), isFalse); + release.complete(); + await expectLater(pending, throwsA(isA())); + final clock = session.evidenceClock; + expect( + clock.snapshot(30177, viewer.public, agent.public, [])(), + isTrue, + ); + return; + } + final newer = switch (mode) { + 'policy' => policy('nobody', 200), + 'runtime' => signed(agent, 10100, { + 'respond_to': 'nobody', + }, time: 200), + 'missing profile' => profile(agent, [ + authTag(viewer, agent.public), + ], createdAt: 200), + 'roster' => roster(200), + _ => profile(other, [], createdAt: 200), + }; + // Same production socket receipt handler used by discovery/user-cache + // streams. It runs immediately, before the 16ms UI batch or query finish. + if (queued) { + release.complete(); + final evidence = await pending; + gate.activate(10); + final publication = withRelayPublicationGuard( + () { + if (!evidence.values.every((e) => e.isCurrent())) { + throw StateError('observed evidence changed before enqueue'); + } + }, + () => SignedEventRelay(session: session, nsec: viewer.nsec).submit( + kind: 9000, + content: '', + tags: [ + ['h', 'room'], + ['p', agent.public], + ], + ), + ); + session.debugHandleMessage(['EVENT', subId, newer.toJson()]); + gate.reset(); + await expectLater(publication, throwsA(isA())); + expect(socket.messages.where((p) => p.first == 'EVENT'), isEmpty); + session.debugFlushEventBuffer(); + return; + } + session.debugHandleMessage(['EVENT', subId, newer.toJson()]); + release.complete(); + if (mode == 'unrelated') { + expect((await pending).keys, {agent.public}); + } else { + await expectLater(pending, throwsA(isA())); + } + }, + ); + } +} + +class _Config extends RelayConfigNotifier { + _Config(this.key); + final String key; + @override + RelayConfig build() => + RelayConfig(baseUrl: 'https://relay.example', nsec: key); +} + +class _Socket extends RelaySocket { + _Socket() + : super( + wsUrl: 'wss://relay.example', + nsec: null, + onMessage: (_) {}, + onConnected: () {}, + onDisconnected: (_) {}, + ); + final messages = >[]; + @override + void send(List payload) => messages.add(payload); +} + +class _Auth extends AuthNotifier { + @override + Future build() async => + const AuthState(status: AuthStatus.unauthenticated); +}