From 1eabb049c6f36f7a5296dc82f0895ce4326bb6dd Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Wed, 9 Sep 2026 12:59:55 -0400 Subject: [PATCH] fix(mobile): recover discovery subscriptions and coordinate deletions Signed-off-by: Logan Johnson --- .../lib/shared/mentions/agent_discovery.dart | 52 +++++++-- .../mentions/agent_identity_provider.dart | 2 + .../shared/mentions/agent_discovery_test.dart | 100 ++++++++++++++++++ 3 files changed, 147 insertions(+), 7 deletions(-) diff --git a/mobile/lib/shared/mentions/agent_discovery.dart b/mobile/lib/shared/mentions/agent_discovery.dart index e54a3fd732..203335f7c2 100644 --- a/mobile/lib/shared/mentions/agent_discovery.dart +++ b/mobile/lib/shared/mentions/agent_discovery.dart @@ -51,18 +51,23 @@ Future> _ownedAgentKeys( // Separate subscription owner: a refresh must not tear down/replay its trigger. class _AgentDirectoryUpdates extends Notifier { + Object? failure; @override int build() { final sessionState = ref.watch(relaySessionProvider); ref.watch(myPubkeyProvider); ref.watch(relayConfigProvider); + failure = null; var disposed = false; + var attempts = 0; + Timer? retry; void Function()? unsubscribe; Timer? debounce; ref.onDispose(() { disposed = true; unsubscribe?.call(); debounce?.cancel(); + retry?.cancel(); }); void changed() { if (disposed || debounce != null) return; @@ -74,17 +79,26 @@ class _AgentDirectoryUpdates extends Notifier { if (sessionState.status == SessionStatus.connected) { final session = ref.read(relaySessionProvider.notifier); - unawaited(() async { + Future subscribe() async { + if (disposed) return; + attempts++; try { final close = await session.subscribeWithStatus( NostrFilter( - kinds: const [0, 10100, 30177, 39002], + kinds: const [0, 5, 10100, 30177, 39002], limit: 0, since: DateTime.now().millisecondsSinceEpoch ~/ 1000, ), - (_) => changed(), + (event) { + if (event.kind == 5 && !_isAgentCoordinateDeletion(event)) return; + changed(); + }, onClosed: (_) => changed(), - onStatusChanged: (_) => changed(), + onStatusChanged: (status) { + if (disposed) return; + if (status == RelaySubscriptionStatus.ready) failure = null; + changed(); + }, ); if (disposed) { close(); @@ -92,10 +106,22 @@ class _AgentDirectoryUpdates extends Notifier { unsubscribe = close; } } catch (error) { - // Suggestions may age during failure; publication always reads fresh. - debugPrint('[AgentDirectory] refresh subscription failed: $error'); + if (disposed) return; + failure = error; + changed(); + // Establishment can fail before the session installs status callbacks. + // Three attempts per session generation; exhausted failure stays + // visible to directory readers until reconnect/rebuild permits retry. + if (attempts < 3) { + retry = Timer(Duration(milliseconds: 250 * attempts), () { + retry = null; + unawaited(subscribe()); + }); + } } - }()); + } + + unawaited(subscribe()); } return 0; } @@ -103,3 +129,15 @@ class _AgentDirectoryUpdates extends Notifier { final _agentDirectoryUpdatesProvider = NotifierProvider<_AgentDirectoryUpdates, int>(_AgentDirectoryUpdates.new); + +// Desktop build_agent_delete emits a kind:5 a-coordinate, not a 30177 event. +bool _isAgentCoordinateDeletion(NostrEvent event) => + verifySignedEvent(event) && + event.tags.any((tag) { + if (tag.length < 2 || tag[0] != 'a') return false; + final coordinate = tag[1].split(':'); + return coordinate.length == 3 && + coordinate[0] == '30177' && + coordinate[1] == event.pubkey && + RegExp(r'^[0-9a-f]{64}$').hasMatch(coordinate[2]); + }); diff --git a/mobile/lib/shared/mentions/agent_identity_provider.dart b/mobile/lib/shared/mentions/agent_identity_provider.dart index fd5cd7e79a..89c1c9d75c 100644 --- a/mobile/lib/shared/mentions/agent_identity_provider.dart +++ b/mobile/lib/shared/mentions/agent_identity_provider.dart @@ -74,6 +74,8 @@ final agentDirectoryProvider = FutureProvider>(( ref, ) async { ref.watch(_agentDirectoryUpdatesProvider); + final failure = ref.read(_agentDirectoryUpdatesProvider.notifier).failure; + if (failure != null) throw failure; final sessionState = ref.watch(relaySessionProvider); if (sessionState.status != SessionStatus.connected) return const []; final session = ref.read(relaySessionProvider.notifier); diff --git a/mobile/test/shared/mentions/agent_discovery_test.dart b/mobile/test/shared/mentions/agent_discovery_test.dart index 94558449ad..9e5aa5a3db 100644 --- a/mobile/test/shared/mentions/agent_discovery_test.dart +++ b/mobile/test/shared/mentions/agent_discovery_test.dart @@ -13,6 +13,13 @@ class _Session extends PolicySession { void Function(NostrEvent)? changed; void Function(RelaySubscriptionStatus)? status; int closed = 0; + int attempts = 0; + int failures = 0; + NostrFilter? subscription; + void deliver(NostrEvent event) { + if (subscription?.kinds.contains(event.kind) == true) changed?.call(event); + } + List Function(NostrFilter)? ownedPage; @override Future fetchRelaySelf() async => authority; @@ -23,8 +30,12 @@ class _Session extends PolicySession { void Function(String)? onClosed, required void Function(RelaySubscriptionStatus) onStatusChanged, }) async { + attempts++; + if (attempts <= failures) throw StateError('establishment unavailable'); + subscription = filter; changed = onEvent; status = onStatusChanged; + onStatusChanged(RelaySubscriptionStatus.ready); return () { closed++; }; @@ -151,6 +162,95 @@ void main() { }, ); + test( + 'failed establishment recovers while connected; coordinate delete refreshes', + () async { + final events = [owned, policy]; + final session = _Session(events, relay.public)..failures = 1; + final c = container(session); + addTearDown(c.dispose); + await c.read(agentDirectoryProvider.future); + await Future.delayed(const Duration(milliseconds: 180)); + await expectLater( + c.read(agentDirectoryProvider.future), + throwsStateError, + ); + await Future.delayed(const Duration(milliseconds: 300)); + expect(session.attempts, 2); + expect((await c.read(agentDirectoryProvider.future)).length, 1); + final deny = signed( + owner, + 30177, + {'name': 'Remote helper', 'parallelism': 1, 'respond_to': 'nobody'}, + time: 101, + tags: [ + ['d', agent.public], + ], + ); + events.add(deny); + session.deliver(deny); + await Future.delayed(const Duration(milliseconds: 180)); + expect( + (await c.read(agentDirectoryProvider.future)).single.respondTo, + 'nobody', + ); + final deletion = signed( + owner, + 5, + '', + tags: [ + ['a', '30177:${owner.public}:${agent.public}'], + ], + ); + events.remove(policy); + events.remove(deny); + session.deliver( + signed( + owner, + 5, + '', + tags: [ + ['a', '30000:${owner.public}:other'], + ], + ), + ); + await Future.delayed(const Duration(milliseconds: 180)); + expect( + (await c.read(agentDirectoryProvider.future)).single.respondTo, + 'nobody', + ); + session.deliver(deletion); + await Future.delayed(const Duration(milliseconds: 180)); + expect(await c.read(agentDirectoryProvider.future), isEmpty); + session.status!(RelaySubscriptionStatus.ready); + session.deliver(deletion); // replay remains removed after reconnect + await Future.delayed(const Duration(milliseconds: 180)); + expect(await c.read(agentDirectoryProvider.future), isEmpty); + }, + ); + + test( + 'establishment retries are bounded and retired scopes cannot retry', + () async { + final session = _Session([owned, policy], relay.public)..failures = 99; + final c = container(session); + await c.read(agentDirectoryProvider.future); + await Future.delayed(const Duration(milliseconds: 1100)); + expect(session.attempts, 3); + await expectLater( + c.read(agentDirectoryProvider.future), + throwsStateError, + ); + c.dispose(); + final retired = _Session([owned, policy], relay.public)..failures = 99; + final c2 = container(retired); + await c2.read(agentDirectoryProvider.future); + c2.dispose(); + await Future.delayed(const Duration(milliseconds: 300)); + expect(retired.attempts, 1); + }, + ); + test( 'owned coordinates paginate equal-time pages and reject a stalled cursor', () async {