fix(mobile): read fresh exact-key agent authorization and relay membership

Signed-off-by: Logan Johnson <loganj@squareup.com>

mobile/lib/shared/mentions/agent_authorization.dart

@@ -0,0 +1,133 @@
+part of 'agent_identity_provider.dart';
+
+/// Fresh exact-key policy and relay-signed membership. Directory entries are
+/// hints; only this read may authorize agent mentions at a destination.
+Future<List<AgentDirectoryEntry>> readAgentAuthorization(
+  RelaySessionNotifier session,
+  Set<String> requestedKeys, {
+  required String? viewer,
+  String? channelId,
+  required bool Function() isCurrent,
+}) async {
+  void check() {
+    if (!isCurrent()) throw StateError('Agent authorization scope changed');
+  }
+
+  check();
+  if (requestedKeys.isEmpty) return [];
+  if (viewer == null ||
+      requestedKeys.length > 1000 ||
+      requestedKeys.any((key) => !RegExp(r'^[0-9a-f]{64}$').hasMatch(key))) {
+    throw StateError('Invalid agent authorization request');
+  }
+  final authority = await session.fetchRelaySelf();
+  check();
+  final membership = await _membershipPages(
+    session,
+    authority,
+    viewer,
+    channelId,
+    check,
+  );
+  check();
+  final runtime = await _queryAgentFilters(session, [
+    for (final key in requestedKeys)
+      NostrFilter(kinds: const [10100], authors: [key], limit: 1),
+  ], checkCurrent: check);
+  check();
+  // Guard every query stage, not merely the result: a mutable session must not
+  // continue an old account/community request against the newly selected relay.
+  final agents = await resolveAgentPolicies(
+    session,
+    runtime.where((e) => requestedKeys.contains(e.pubkey)).toList(),
+    requestedKeys: requestedKeys,
+    checkCurrent: check,
+  );
+  check();
+  final latest = <String, NostrEvent>{};
+  for (final event in membership) {
+    if (event.kind != 39002 ||
+        event.pubkey != authority ||
+        !verifySignedEvent(event)) {
+      continue;
+    }
+    final destination = event.getTagValue('d');
+    if (destination == null ||
+        (channelId != null && destination != channelId)) {
+      continue;
+    }
+    final previous = latest[destination];
+    if (previous == null || _newer(event, previous)) {
+      latest[destination] = event;
+    }
+  }
+  final result = <AgentDirectoryEntry>[];
+  for (final agent in agents) {
+    final owned = agent.ownerPubkey == viewer;
+    final channels = <String>[];
+    for (final entry in latest.entries) {
+      final people = entry.value.tags.where(
+        (tag) => tag.length >= 2 && tag[0] == 'p',
+      );
+      if (!people.any((tag) => tag[1] == viewer)) continue;
+      if (people.any(
+        (tag) =>
+            tag[1] == agent.pubkey &&
+            (owned || (tag.length >= 4 && tag[3] == 'bot')),
+      )) {
+        channels.add(entry.key);
+      }
+    }
+    if (!owned && channels.isEmpty) continue;
+    result.add(
+      AgentDirectoryEntry(
+        pubkey: agent.pubkey,
+        ownerPubkey: agent.ownerPubkey,
+        displayName: agent.displayName,
+        respondTo: agent.respondTo,
+        respondToAllowlist: agent.respondToAllowlist,
+        channelIds: channels,
+      ),
+    );
+  }
+  return result;
+}
+
+Future<List<NostrEvent>> _membershipPages(
+  RelaySessionNotifier session,
+  String authority,
+  String viewer,
+  String? channelId,
+  void Function() check,
+) async {
+  final result = <NostrEvent>[];
+  NostrEvent? cursor;
+  for (var page = 0; page < 200; page++) {
+    check();
+    final events = await session.queryRelay([
+      NostrFilter(
+        kinds: const [39002],
+        authors: [authority],
+        tags: {
+          '#p': [viewer],
+          if (channelId != null) '#d': [channelId],
+        },
+        limit: 500,
+        until: cursor?.createdAt,
+        extensions: {if (cursor != null) 'before_id': cursor.id},
+      ),
+    ]);
+    check();
+    result.addAll(events);
+    if (events.length < 500) return result;
+    final next = events.last;
+    if (cursor != null &&
+        (next.createdAt > cursor.createdAt ||
+            (next.createdAt == cursor.createdAt &&
+                next.id.compareTo(cursor.id) <= 0))) {
+      throw StateError('Membership pagination did not advance');
+    }
+    cursor = next;
+  }
+  throw StateError('Membership query exceeded its page budget');
+}

mobile/lib/shared/mentions/agent_identity_provider.dart

@@ -9,6 +9,7 @@ import '../../shared/crypto/signed_event.dart';
 import '../../shared/relay/relay.dart';
 
 part 'agent_policy.dart';
+part 'agent_authorization.dart';
 
 /// A relay agent parsed from its kind:10100 agent-profile event.
 ///
@@ -70,8 +71,19 @@ final agentDirectoryProvider = FutureProvider<List<AgentDirectoryEntry>>((
   final sessionState = ref.watch(relaySessionProvider);
   if (sessionState.status != SessionStatus.connected) return const [];
   final session = ref.read(relaySessionProvider.notifier);
+  final config = ref.read(relayConfigProvider);
+  var disposed = false;
+  ref.onDispose(() => disposed = true);
+  final viewer = ref.read(myPubkeyProvider);
   final events = await session.fetchHistory(NostrFilters.agentProfiles());
-  return resolveAgentPolicies(session, events);
+  if (disposed) return [];
+  return readAgentAuthorization(
+    session,
+    events.map((event) => event.pubkey).toSet(),
+    viewer: viewer,
+    isCurrent: () =>
+        !disposed && identical(config, ref.read(relayConfigProvider)),
+  );
 });
 
 /// Verified NIP-OA owner pubkey per agent pubkey, from the agents' kind:0

mobile/lib/shared/mentions/agent_policy.dart

@@ -9,10 +9,12 @@ bool _newer(NostrEvent event, NostrEvent previous) =>
 /// is distinct from a failed read; failures must never revive runtime access.
 Future<List<NostrEvent>> _queryAgentFilters(
   RelaySessionNotifier session,
-  List<NostrFilter> filters,
-) async {
+  List<NostrFilter> filters, {
+  void Function()? checkCurrent,
+}) async {
   final events = <NostrEvent>[];
   for (var start = 0; start < filters.length; start += 10) {
+    checkCurrent?.call();
     events.addAll(
       await session.queryRelay(filters.skip(start).take(10).toList()),
     );
@@ -24,8 +26,10 @@ Future<List<NostrEvent>> _queryAgentFilters(
 /// directory. This does not expand discovery to owner-only coordinates yet.
 Future<List<AgentDirectoryEntry>> resolveAgentPolicies(
   RelaySessionNotifier session,
-  List<NostrEvent> runtimeEvents,
-) async {
+  List<NostrEvent> runtimeEvents, {
+  Set<String>? requestedKeys,
+  void Function()? checkCurrent,
+}) async {
   final latest = <String, NostrEvent>{};
   for (final event in runtimeEvents.where((event) => event.kind == 10100)) {
     final previous = latest[event.pubkey];
@@ -33,19 +37,21 @@ Future<List<AgentDirectoryEntry>> resolveAgentPolicies(
       latest[event.pubkey] = event;
     }
   }
+  final keys = requestedKeys ?? latest.keys.toSet();
   final profiles = latestProfileEvents(
     await _queryAgentFilters(session, [
-      for (final key in latest.keys)
+      for (final key in keys)
         NostrFilter(kinds: const [0], authors: [key], limit: 1),
-    ]),
+    ], checkCurrent: checkCurrent),
   );
   final owners = <String, String>{};
   for (final profile in profiles.values) {
     final owner = verifiedOaOwnerPubkey(profile);
-    if (owner != null && latest.containsKey(profile.pubkey)) {
+    if (owner != null && keys.contains(profile.pubkey)) {
       owners[profile.pubkey] = owner;
     }
   }
+  checkCurrent?.call();
   final policies = await _queryAgentFilters(session, [
     for (final owner in owners.entries)
       NostrFilter(
@@ -56,7 +62,8 @@ Future<List<AgentDirectoryEntry>> resolveAgentPolicies(
         },
         limit: 1,
       ),
-  ]);
+  ], checkCurrent: checkCurrent);
+  checkCurrent?.call();
   return mergeAgentPolicies(latest.values, policies, owners);
 }
 

mobile/lib/shared/relay/relay_http_query_client.dart

@@ -15,6 +15,29 @@ class RelayHttpQueryClient {
   _ClientGeneration? _currentGeneration;
   final Set<_ClientGeneration> _generations = {};
 
+  /// Read relay metadata over the same owned transport as query requests.
+  Future<http.Response> get(
+    Uri url, {
+    required Map<String, String> headers,
+    required Duration timeout,
+  }) async {
+    final generation = _injectedClient == null
+        ? (_currentGeneration ??= _createGeneration())
+        : null;
+    generation?.acquire();
+    try {
+      return await (_injectedClient ?? generation!.client)
+          .get(url, headers: headers)
+          .timeout(timeout);
+    } on TimeoutException {
+      if (identical(_currentGeneration, generation)) _currentGeneration = null;
+      generation?.retire();
+      rethrow;
+    } finally {
+      generation?.release();
+    }
+  }
+
   Future<http.Response> post(
     Uri url, {
     required Map<String, String> headers,

mobile/lib/shared/relay/relay_session.dart

@@ -149,6 +149,25 @@ class RelaySessionNotifier extends Notifier<SessionState> {
     return const SessionState(status: SessionStatus.disconnected);
   }
 
+  /// NIP-11 `self` is the membership signer; `pubkey` is only contact metadata.
+  Future<String> fetchRelaySelf() async {
+    final config = ref.read(relayConfigProvider);
+    final response = await _httpQueryClient.get(
+      Uri.parse(config.baseUrl),
+      headers: const {'Accept': 'application/nostr+json'},
+      timeout: const Duration(seconds: 8),
+    );
+    if (response.statusCode != 200) {
+      throw RelayException(response.statusCode, 'Relay authority unavailable');
+    }
+    final data = jsonDecode(response.body);
+    final key = data is Map ? data['self'] : null;
+    if (key is! String || !RegExp(r'^[0-9a-fA-F]{64}$').hasMatch(key)) {
+      throw const FormatException('Relay membership authority unavailable');
+    }
+    return key.toLowerCase();
+  }
+
   /// Execute a one-shot query via the relay's HTTP bridge (`POST /query`).
   Future<List<NostrEvent>> queryRelay(
     List<NostrFilter> filters, {

mobile/test/shared/mentions/agent_authorization_test.dart

@@ -0,0 +1,257 @@
+import 'dart:convert';
+
+import 'package:buzz/shared/mentions/agent_identity_provider.dart';
+import 'package:buzz/shared/relay/relay.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 PolicySession, signed;
+
+class _ReadSession extends PolicySession {
+  _ReadSession(super.events, this.authority);
+  final String authority;
+  void Function(List<NostrFilter>)? onQuery;
+  List<NostrEvent> Function(List<NostrFilter>)? page;
+  @override
+  Future<String> fetchRelaySelf() async => authority;
+  @override
+  Future<List<NostrEvent>> queryRelay(
+    List<NostrFilter> filters, {
+    Duration timeout = const Duration(seconds: 8),
+  }) async {
+    onQuery?.call(filters);
+    if (page != null && filters.first.kinds.contains(39002)) {
+      return page!(filters);
+    }
+    return super.queryRelay(filters);
+  }
+}
+
+void main() {
+  final owner = nostr.Keys.generate();
+  final agent = nostr.Keys.generate();
+  final relay = nostr.Keys.generate();
+  final other = nostr.Keys.generate();
+  final owned = profile(agent, [authTag(owner, agent.public)]);
+  final policy = signed(
+    owner,
+    30177,
+    {'name': 'Agent', 'parallelism': 1, 'respond_to': 'anyone'},
+    tags: [
+      ['d', agent.public],
+    ],
+  );
+  final runtime = signed(agent, 10100, {
+    'name': 'Agent',
+    'respond_to': 'anyone',
+    'channel_ids': ['forged-channel'],
+  });
+  NostrEvent members({
+    String channel = 'room',
+    nostr.Keys? signer,
+    int time = 100,
+    bool includeAgent = true,
+    bool includeViewer = true,
+  }) => signed(
+    signer ?? relay,
+    39002,
+    '',
+    time: time,
+    tags: [
+      ['d', channel],
+      if (includeViewer) ['p', owner.public],
+      if (includeAgent) ['p', agent.public, '', 'member'],
+    ],
+  );
+
+  Future<List<AgentDirectoryEntry>> read(
+    _ReadSession session, {
+    bool Function()? isCurrent,
+    String? destination = 'room',
+  }) async {
+    final container = ProviderContainer(
+      overrides: [relaySessionProvider.overrideWith(() => session)],
+    );
+    container.read(relaySessionProvider);
+    try {
+      return await readAgentAuthorization(
+        session,
+        {agent.public},
+        viewer: owner.public,
+        channelId: destination,
+        isCurrent: isCurrent ?? () => true,
+      );
+    } finally {
+      container.dispose();
+    }
+  }
+
+  test(
+    'owned agent needs no runtime and ordinary membership role is sufficient',
+    () async {
+      final session = _ReadSession([owned, policy, members()], relay.public);
+      final result = await read(session);
+      expect(result.single.channelIds, ['room']);
+      final filter = session.queries.first;
+      expect(filter.authors, [relay.public]);
+      expect(filter.tags, {
+        '#p': [owner.public],
+        '#d': ['room'],
+      });
+      expect(
+        session.queries.where((q) => q.kinds.contains(10100)).single.authors,
+        [agent.public],
+      );
+    },
+  );
+
+  test(
+    'runtime cannot assert membership; exact destination excludes other rooms',
+    () async {
+      final result = await read(
+        _ReadSession([
+          owned,
+          policy,
+          runtime,
+          members(channel: 'elsewhere'),
+        ], relay.public),
+      );
+      expect(result.single.channelIds, isEmpty);
+    },
+  );
+
+  test(
+    'only relay self signature is authority; latest removals do not fall back',
+    () async {
+      for (final snapshots in [
+        [members(signer: other)],
+        [
+          NostrEvent.fromJson({...members().toJson(), 'content': 'tampered'}),
+        ],
+        [members(), members(time: 101, includeAgent: false)],
+        [members(), members(time: 101, includeViewer: false)],
+      ]) {
+        final result = await read(
+          _ReadSession([owned, policy, ...snapshots], relay.public),
+        );
+        expect(result.single.channelIds, isEmpty);
+      }
+    },
+  );
+
+  test('same-second membership tie is deterministic', () async {
+    final yes = members();
+    final no = members(includeAgent: false);
+    for (final snapshots in [
+      [yes, no],
+      [no, yes],
+    ]) {
+      final result = await read(
+        _ReadSession([owned, policy, ...snapshots], relay.public),
+      );
+      expect(result.single.channelIds.isNotEmpty, yes.id.compareTo(no.id) < 0);
+    }
+  });
+
+  test('scope changes stop subsequent policy queries and results', () async {
+    var current = true;
+    final session = _ReadSession([owned, policy, members()], relay.public);
+    session.onQuery = (filters) {
+      if (filters.any((f) => f.kinds.contains(0))) current = false;
+    };
+    await expectLater(
+      read(session, isCurrent: () => current),
+      throwsStateError,
+    );
+    expect(session.queries.any((q) => q.kinds.contains(30177)), isFalse);
+  });
+
+  test(
+    'membership failure propagates, not an authoritative empty result',
+    () async {
+      final session = _ReadSession([owned, policy], relay.public);
+      session.onQuery = (filters) {
+        if (filters.first.kinds.contains(39002)) {
+          throw StateError('query unavailable');
+        }
+      };
+      await expectLater(read(session), throwsStateError);
+    },
+  );
+
+  test(
+    'membership pagination advances equal-time cursor and rejects stalled pages',
+    () async {
+      final first = members(channel: 'a');
+      final last = members(channel: 'z');
+      final session = _ReadSession([owned, policy], relay.public);
+      var calls = 0;
+      session.page = (filters) {
+        calls++;
+        if (calls == 1) return List.filled(500, first);
+        expect(filters.single.until, first.createdAt);
+        expect(filters.single.extensions['before_id'], first.id);
+        return [last];
+      };
+      final result = await read(session, destination: null);
+      expect(result.single.channelIds.toSet(), {'a', 'z'});
+      calls = 0;
+      final stalled = _ReadSession([owned, policy], relay.public)
+        ..page = (_) => List.filled(500, first);
+      await expectLater(read(stalled), throwsStateError);
+    },
+  );
+
+  test(
+    'NIP11 uses self, preserves host and propagates unavailable authority',
+    () async {
+      var body = jsonEncode({
+        'self': relay.public.toUpperCase(),
+        'pubkey': other.public,
+      });
+      var status = 200;
+      final client = MockClient((request) async {
+        expect(request.url.toString(), 'https://tenant.example/');
+        expect(request.headers['Accept'], 'application/nostr+json');
+        return http.Response(body, status);
+      });
+      final container = ProviderContainer(
+        overrides: [
+          relayConfigProvider.overrideWith(_Config.new),
+          relaySessionProvider.overrideWith(
+            () => RelaySessionNotifier(httpClient: client),
+          ),
+        ],
+      );
+      try {
+        final session = container.read(relaySessionProvider.notifier);
+        expect(await session.fetchRelaySelf(), relay.public);
+        for (final invalid in [
+          jsonEncode({'pubkey': relay.public}),
+          '{}',
+          'not json',
+          '{"self":"bad"}',
+        ]) {
+          body = invalid;
+          await expectLater(session.fetchRelaySelf(), throwsFormatException);
+        }
+        status = 503;
+        await expectLater(
+          session.fetchRelaySelf(),
+          throwsA(isA<RelayException>()),
+        );
+      } finally {
+        container.dispose();
+      }
+    },
+  );
+}
+
+class _Config extends RelayConfigNotifier {
+  @override
+  RelayConfig build() => const RelayConfig(baseUrl: 'https://tenant.example/');
+}

mobile/test/shared/mentions/agent_policy_test.dart

@@ -33,6 +33,9 @@ class PolicySession extends RelaySessionNotifier {
   final bool failPolicy;
   final queries = <NostrFilter>[];
   @override
+  Future<String> fetchRelaySelf() async =>
+      events.firstWhere((e) => e.kind == 39002).pubkey;
+  @override
   SessionState build() => const SessionState(status: SessionStatus.connected);
   @override
   Future<List<NostrEvent>> fetchHistory(
@@ -94,9 +97,24 @@ void main() {
     bool failPolicy = false,
     void Function(PolicySession)? inspect,
   }) async {
-    final session = PolicySession(events, failPolicy: failPolicy);
+    final session = PolicySession([
+      ...events,
+      signed(
+        owner,
+        39002,
+        '',
+        tags: [
+          ['d', 'channel'],
+          ['p', owner.public],
+          ['p', agent.public, '', 'bot'],
+        ],
+      ),
+    ], failPolicy: failPolicy);
     final container = ProviderContainer(
-      overrides: [relaySessionProvider.overrideWith(() => session)],
+      overrides: [
+        relaySessionProvider.overrideWith(() => session),
+        myPubkeyProvider.overrideWithValue(owner.public),
+      ],
     );
     try {
       final result = await container.read(agentDirectoryProvider.future);