[codex] Add mobile agent activity viewer (#436)

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

mobile/test/features/channels/agent_activity/observer_subscription_test.dart

@@ -0,0 +1,375 @@
+import 'dart:async';
+import 'dart:convert';
+
+import 'package:flutter_test/flutter_test.dart';
+import 'package:hooks_riverpod/hooks_riverpod.dart';
+import 'package:nostr/nostr.dart' as nostr;
+import 'package:sprout_mobile/features/channels/agent_activity/observer_models.dart';
+import 'package:sprout_mobile/features/channels/agent_activity/observer_subscription.dart';
+import 'package:sprout_mobile/shared/crypto/nip44.dart';
+import 'package:sprout_mobile/shared/relay/relay.dart';
+
+void main() {
+  test('provider initializes without circular dependency error', () {
+    // Regression test: reading the provider should NOT throw
+    // "Bad state: Tried to read the state of an uninitialized provider".
+    final container = ProviderContainer(
+      overrides: [
+        relaySessionProvider.overrideWith(() => _RecordingRelaySession()),
+        relayConfigProvider.overrideWith(
+          () => _FakeRelayConfigNotifier(nsec: null),
+        ),
+      ],
+    );
+    addTearDown(container.dispose);
+
+    const key = (channelId: 'test-channel', agentPubkey: 'deadbeef');
+    // This line threw before the fix.
+    final state = container.read(observerSubscriptionProvider(key));
+    expect(state.connection, ObserverConnectionState.idle);
+    expect(state.transcript, isEmpty);
+  });
+
+  test('transitions to error when nsec is invalid', () async {
+    final container = ProviderContainer(
+      overrides: [
+        relaySessionProvider.overrideWith(() => _RecordingRelaySession()),
+        relayConfigProvider.overrideWith(
+          () => _FakeRelayConfigNotifier(nsec: 'nsec1invalid'),
+        ),
+      ],
+    );
+    addTearDown(container.dispose);
+
+    const key = (channelId: 'test-channel', agentPubkey: 'deadbeef');
+    container.read(observerSubscriptionProvider(key));
+
+    // Let the subscription microtask run.
+    await Future<void>.delayed(Duration.zero);
+
+    final state = container.read(observerSubscriptionProvider(key));
+    // With invalid nsec, it should be in error state, NOT throw.
+    expect(state.connection, ObserverConnectionState.error);
+    expect(state.errorMessage, isNotNull);
+  });
+
+  test('stays idle when nsec is null', () async {
+    final container = ProviderContainer(
+      overrides: [
+        relaySessionProvider.overrideWith(() => _RecordingRelaySession()),
+        relayConfigProvider.overrideWith(
+          () => _FakeRelayConfigNotifier(nsec: null),
+        ),
+      ],
+    );
+    addTearDown(container.dispose);
+
+    const key = (channelId: 'test-channel', agentPubkey: 'deadbeef');
+    container.read(observerSubscriptionProvider(key));
+
+    // Let the subscription microtask run. Without an nsec, it should no-op.
+    await Future<void>.delayed(Duration.zero);
+
+    final state = container.read(observerSubscriptionProvider(key));
+    expect(state.connection, ObserverConnectionState.idle);
+    expect(state.transcript, isEmpty);
+  });
+
+  test(
+    'subscribes with correct filter shape and transitions to open',
+    () async {
+      final userKeychain = nostr.Keychain.generate();
+      final nsec = nostr.Nip19.encodePrivkey(userKeychain.private);
+      final myPubkey = userKeychain.public;
+      // Agent needs a valid 64-char hex pubkey for getConversationKey.
+      final agentKeychain = nostr.Keychain.generate();
+      final agentPubkey = agentKeychain.public;
+
+      final relaySession = _RecordingRelaySession();
+      final container = ProviderContainer(
+        overrides: [
+          relaySessionProvider.overrideWith(() => relaySession),
+          relayConfigProvider.overrideWith(
+            () => _FakeRelayConfigNotifier(nsec: nsec),
+          ),
+        ],
+      );
+      addTearDown(container.dispose);
+
+      final key = (channelId: 'test-channel', agentPubkey: agentPubkey);
+      container.read(observerSubscriptionProvider(key));
+
+      // Let the subscription microtask run.
+      await Future<void>.delayed(Duration.zero);
+
+      final state = container.read(observerSubscriptionProvider(key));
+      expect(state.connection, ObserverConnectionState.open);
+      expect(state.transcript, isEmpty);
+
+      // Verify the subscription filter shape.
+      expect(relaySession.filters, hasLength(1));
+      final filter = relaySession.filters.first;
+      expect(filter.kinds, [EventKind.agentObserverFrame]);
+      expect(filter.limit, 0);
+      expect(filter.tags['#p'], contains(myPubkey));
+      expect(filter.since, isNull);
+    },
+  );
+
+  test(
+    'uses one shared relay subscription for channel-scoped readers',
+    () async {
+      final userKeychain = nostr.Keychain.generate();
+      final agentKeychain = nostr.Keychain.generate();
+      final relaySession = _RecordingRelaySession();
+      final container = ProviderContainer(
+        overrides: [
+          relaySessionProvider.overrideWith(() => relaySession),
+          relayConfigProvider.overrideWith(
+            () => _FakeRelayConfigNotifier(
+              nsec: nostr.Nip19.encodePrivkey(userKeychain.private),
+            ),
+          ),
+        ],
+      );
+      addTearDown(container.dispose);
+
+      container.read(
+        observerSubscriptionProvider((
+          channelId: 'first-channel',
+          agentPubkey: agentKeychain.public,
+        )),
+      );
+      container.read(
+        observerSubscriptionProvider((
+          channelId: 'second-channel',
+          agentPubkey: agentKeychain.public,
+        )),
+      );
+
+      await Future<void>.delayed(Duration.zero);
+
+      expect(relaySession.filters, hasLength(1));
+    },
+  );
+
+  test('surfaces relay CLOSED messages through observer state', () async {
+    final userKeychain = nostr.Keychain.generate();
+    final agentKeychain = nostr.Keychain.generate();
+    final relaySession = _RecordingRelaySession();
+    final container = ProviderContainer(
+      overrides: [
+        relaySessionProvider.overrideWith(() => relaySession),
+        relayConfigProvider.overrideWith(
+          () => _FakeRelayConfigNotifier(
+            nsec: nostr.Nip19.encodePrivkey(userKeychain.private),
+          ),
+        ),
+      ],
+    );
+    addTearDown(container.dispose);
+
+    final key = (channelId: 'test-channel', agentPubkey: agentKeychain.public);
+    container.read(observerSubscriptionProvider(key));
+    await Future<void>.delayed(Duration.zero);
+
+    relaySession.closeAll('restricted: p-gated events require #p');
+
+    final state = container.read(observerSubscriptionProvider(key));
+    expect(state.connection, ObserverConnectionState.error);
+    expect(state.errorMessage, contains('p-gated events require #p'));
+  });
+
+  test('ignores stale subscribe completion after identity changes', () async {
+    final firstUser = nostr.Keychain.generate();
+    final secondUser = nostr.Keychain.generate();
+    final agentKeychain = nostr.Keychain.generate();
+    final relaySession = _RecordingRelaySession()..delaySubscribes = true;
+    final container = ProviderContainer(
+      overrides: [
+        relaySessionProvider.overrideWith(() => relaySession),
+        relayConfigProvider.overrideWith(
+          () => _FakeRelayConfigNotifier(
+            nsec: nostr.Nip19.encodePrivkey(firstUser.private),
+          ),
+        ),
+      ],
+    );
+    addTearDown(container.dispose);
+
+    final key = (channelId: 'test-channel', agentPubkey: agentKeychain.public);
+    container.read(observerSubscriptionProvider(key));
+    await Future<void>.delayed(Duration.zero);
+
+    expect(relaySession.filters, hasLength(1));
+    expect(relaySession.filters.single.tags['#p'], [firstUser.public]);
+
+    (container.read(relayConfigProvider.notifier) as _FakeRelayConfigNotifier)
+        .setNsec(nostr.Nip19.encodePrivkey(secondUser.private));
+    container.read(observerSubscriptionProvider(key));
+    await Future<void>.delayed(Duration.zero);
+
+    expect(relaySession.filters, hasLength(2));
+    expect(relaySession.filters.last.tags['#p'], [secondUser.public]);
+
+    relaySession.releaseSubscribe(0);
+    await Future<void>.delayed(Duration.zero);
+
+    expect(relaySession.filters, hasLength(1));
+    expect(relaySession.filters.single.tags['#p'], [secondUser.public]);
+
+    relaySession.releaseSubscribe(1);
+    await Future<void>.delayed(Duration.zero);
+
+    final state = container.read(observerSubscriptionProvider(key));
+    expect(state.connection, ObserverConnectionState.open);
+    expect(relaySession.filters, hasLength(1));
+  });
+
+  test(
+    'decrypts observer frames and exposes channel-scoped transcript',
+    () async {
+      final ownerKeychain = nostr.Keychain.generate();
+      final agentKeychain = nostr.Keychain.generate();
+      final nsec = nostr.Nip19.encodePrivkey(ownerKeychain.private);
+      final relaySession = _RecordingRelaySession();
+      final container = ProviderContainer(
+        overrides: [
+          relaySessionProvider.overrideWith(() => relaySession),
+          relayConfigProvider.overrideWith(
+            () => _FakeRelayConfigNotifier(nsec: nsec),
+          ),
+        ],
+      );
+      addTearDown(container.dispose);
+
+      const channelId = 'test-channel';
+      final key = (channelId: channelId, agentPubkey: agentKeychain.public);
+      container.read(observerSubscriptionProvider(key));
+      await Future<void>.delayed(Duration.zero);
+
+      final conversationKey = getConversationKey(
+        agentKeychain.private,
+        ownerKeychain.public,
+      );
+      final encrypted = nip44Encrypt(
+        conversationKey,
+        jsonEncode({
+          'seq': 1,
+          'timestamp': '2026-04-30T12:00:00.000Z',
+          'kind': 'turn_started',
+          'channelId': channelId,
+          'turnId': 'turn-1',
+          'payload': {
+            'triggeringEventIds': ['0123456789abcdef'],
+          },
+        }),
+      );
+      final event = nostr.Event.from(
+        kind: EventKind.agentObserverFrame,
+        content: encrypted,
+        tags: [
+          ['p', ownerKeychain.public],
+          ['agent', agentKeychain.public],
+          ['frame', 'telemetry'],
+        ],
+        privkey: agentKeychain.private,
+        verify: false,
+      );
+
+      relaySession.emit(NostrEvent.fromJson(event.toJson()));
+
+      final state = container.read(observerSubscriptionProvider(key));
+      expect(state.connection, ObserverConnectionState.open);
+      expect(state.transcript, hasLength(1));
+      final item = state.transcript.single;
+      expect(item, isA<LifecycleItem>());
+      expect((item as LifecycleItem).title, 'Turn started');
+
+      final otherChannelState = container.read(
+        observerSubscriptionProvider((
+          channelId: 'other-channel',
+          agentPubkey: agentKeychain.public,
+        )),
+      );
+      expect(otherChannelState.transcript, isEmpty);
+    },
+  );
+}
+
+// ---------------------------------------------------------------------------
+// Test helpers
+// ---------------------------------------------------------------------------
+
+class _RecordingRelaySession extends RelaySessionNotifier {
+  final List<NostrFilter> filters = [];
+  final List<void Function(NostrEvent)> _listeners = [];
+  final List<void Function(String message)> _closedListeners = [];
+  final List<Completer<void>> _subscribeGates = [];
+  bool delaySubscribes = false;
+
+  @override
+  SessionState build() => const SessionState(status: SessionStatus.connected);
+
+  @override
+  Future<void Function()> subscribe(
+    NostrFilter filter,
+    void Function(NostrEvent) onEvent, {
+    void Function(String message)? onClosed,
+  }) async {
+    filters.add(filter);
+    _listeners.add(onEvent);
+    if (onClosed != null) {
+      _closedListeners.add(onClosed);
+    }
+    if (delaySubscribes) {
+      final gate = Completer<void>();
+      _subscribeGates.add(gate);
+      await gate.future;
+    }
+    return () {
+      filters.remove(filter);
+      _listeners.remove(onEvent);
+      if (onClosed != null) {
+        _closedListeners.remove(onClosed);
+      }
+    };
+  }
+
+  void emit(NostrEvent event) {
+    for (final listener in List.of(_listeners)) {
+      listener(event);
+    }
+  }
+
+  void closeAll(String message) {
+    for (final listener in List.of(_closedListeners)) {
+      listener(message);
+    }
+    filters.clear();
+    _listeners.clear();
+    _closedListeners.clear();
+  }
+
+  void releaseSubscribe(int index) {
+    final gate = _subscribeGates[index];
+    if (!gate.isCompleted) {
+      gate.complete();
+    }
+  }
+}
+
+class _FakeRelayConfigNotifier extends RelayConfigNotifier {
+  String? _nsec;
+
+  _FakeRelayConfigNotifier({required String? nsec}) : _nsec = nsec;
+
+  @override
+  RelayConfig build() =>
+      RelayConfig(baseUrl: 'http://localhost:3000', nsec: _nsec);
+
+  void setNsec(String? nsec) {
+    _nsec = nsec;
+    state = RelayConfig(baseUrl: 'http://localhost:3000', nsec: _nsec);
+  }
+}

mobile/test/features/channels/agent_activity/transcript_builder_test.dart

@@ -0,0 +1,144 @@
+import 'package:flutter_test/flutter_test.dart';
+import 'package:sprout_mobile/features/channels/agent_activity/observer_models.dart';
+import 'package:sprout_mobile/features/channels/agent_activity/transcript_builder.dart';
+
+void main() {
+  test('aggregates assistant chunks until another item seals the message', () {
+    final items = buildTranscript([
+      _updateFrame(
+        seq: 1,
+        update: {
+          'sessionUpdate': 'agent_message_chunk',
+          'messageId': 'm1',
+          'content': [
+            {'type': 'text', 'text': 'Hello'},
+          ],
+        },
+      ),
+      _updateFrame(
+        seq: 2,
+        update: {
+          'sessionUpdate': 'agent_message_chunk',
+          'messageId': 'm1',
+          'content': [
+            {'type': 'text', 'text': ' world'},
+          ],
+        },
+      ),
+      _updateFrame(
+        seq: 3,
+        update: {
+          'sessionUpdate': 'tool_call',
+          'toolCallId': 'tool-1',
+          'title': 'sleep',
+          'args': {'seconds': 5},
+        },
+      ),
+      _updateFrame(
+        seq: 4,
+        update: {
+          'sessionUpdate': 'agent_message_chunk',
+          'messageId': 'm1',
+          'content': [
+            {'type': 'text', 'text': 'Done'},
+          ],
+        },
+      ),
+    ]);
+
+    expect(items, hasLength(3));
+    expect(items[0], isA<MessageItem>());
+    expect((items[0] as MessageItem).text, 'Hello world');
+    expect(items[1], isA<ToolItem>());
+    expect(items[2], isA<MessageItem>());
+    expect((items[2] as MessageItem).text, 'Done');
+  });
+
+  test('normalizes sprout tool calls and applies result updates', () {
+    final items = buildTranscript([
+      _updateFrame(
+        seq: 1,
+        update: {
+          'sessionUpdate': 'tool_call',
+          'toolCallId': 'send-1',
+          'title': 'Sending message to channel',
+          'status': 'executing',
+          'args': {'content': 'hi'},
+        },
+      ),
+      _updateFrame(
+        seq: 2,
+        update: {
+          'sessionUpdate': 'tool_call_update',
+          'toolCallId': 'send-1',
+          'status': 'completed',
+          'content': [
+            {'type': 'text', 'text': 'posted #activity-test-channel'},
+          ],
+        },
+      ),
+    ]);
+
+    expect(items, hasLength(1));
+    expect(items.single, isA<ToolItem>());
+    final tool = items.single as ToolItem;
+    expect(tool.sproutToolName, 'send_message');
+    expect(tool.toolName, 'send_message');
+    expect(tool.status, ToolStatus.completed);
+    expect(tool.args, {'content': 'hi'});
+    expect(tool.result, 'posted #activity-test-channel');
+  });
+
+  test('parses sprout prompt text into user message and metadata', () {
+    final items = buildTranscript([
+      ObserverFrame(
+        seq: 1,
+        timestamp: _timestamp(1),
+        kind: 'acp_write',
+        turnId: 'turn-1',
+        payload: {
+          'method': 'session/prompt',
+          'params': {
+            'prompt': [
+              {
+                'content':
+                    '[Sprout event: stream message]\n'
+                    'Content: @claude can you do that again?\n\n'
+                    '[Channel]\n'
+                    '#activity-test-channel',
+              },
+            ],
+          },
+        },
+      ),
+    ]);
+
+    expect(items, hasLength(2));
+    expect(items[0], isA<MessageItem>());
+    final message = items[0] as MessageItem;
+    expect(message.role, 'user');
+    expect(message.title, 'Stream Message');
+    expect(message.text, '@claude can you do that again?');
+    expect(items[1], isA<MetadataItem>());
+    expect((items[1] as MetadataItem).sections, hasLength(2));
+  });
+}
+
+ObserverFrame _updateFrame({
+  required int seq,
+  required Map<String, dynamic> update,
+}) {
+  return ObserverFrame(
+    seq: seq,
+    timestamp: _timestamp(seq),
+    kind: 'acp_read',
+    turnId: 'turn-1',
+    payload: {
+      'method': 'session/update',
+      'params': {'update': update},
+    },
+  );
+}
+
+String _timestamp(int seq) =>
+    DateTime.utc(2026, 4, 30, 12, 0, seq).toIso8601String();

mobile/test/features/channels/channel_messages_provider_test.dart

@@ -144,8 +144,9 @@ class _RecordingRelaySessionNotifier extends RelaySessionNotifier {
   @override
   Future<void Function()> subscribe(
     NostrFilter filter,
-    void Function(NostrEvent) onEvent,
-  ) async {
+    void Function(NostrEvent) onEvent, {
+    void Function(String message)? onClosed,
+  }) async {
     operations.add('subscribe');
     liveFilters.add(filter);
     if (!_subscribed.isCompleted) {

mobile/test/features/channels/channels_provider_test.dart

@@ -133,8 +133,9 @@ class _RecordingRelaySessionNotifier extends RelaySessionNotifier {
   @override
   Future<void Function()> subscribe(
     NostrFilter filter,
-    void Function(NostrEvent) onEvent,
-  ) async {
+    void Function(NostrEvent) onEvent, {
+    void Function(String message)? onClosed,
+  }) async {
     filters.add(filter);
     _listeners.add(onEvent);
     return () {

mobile/test/shared/relay/relay_session_test.dart

@@ -39,6 +39,57 @@ void main() {
     unsubscribeFirst();
     unsubscribeSecond();
   });
+
+  test('live subscribe fails when relay closes before ready', () async {
+    final session = RelaySessionNotifier();
+    const filter = NostrFilter(kinds: [EventKind.agentObserverFrame], limit: 0);
+
+    final subscribe = session.subscribe(filter, (_) {});
+    session.debugHandleMessage([
+      'CLOSED',
+      'l-1',
+      'restricted: p-gated events require #p matching your pubkey',
+    ]);
+
+    await expectLater(
+      subscribe,
+      throwsA(
+        isA<Exception>().having(
+          (error) => error.toString(),
+          'message',
+          contains('p-gated events require #p'),
+        ),
+      ),
+    );
+  });
+
+  test(
+    'live onClosed callback runs when relay closes an open subscription',
+    () async {
+      final session = RelaySessionNotifier();
+      final closedMessages = <String>[];
+      const filter = NostrFilter(
+        kinds: [EventKind.agentObserverFrame],
+        limit: 0,
+      );
+
+      final subscribe = session.subscribe(
+        filter,
+        (_) {},
+        onClosed: closedMessages.add,
+      );
+      session.debugHandleMessage(['EOSE', 'l-1']);
+      final unsubscribe = await subscribe;
+      session.debugHandleMessage([
+        'CLOSED',
+        'l-1',
+        'restricted: no longer valid',
+      ]);
+
+      expect(closedMessages, ['restricted: no longer valid']);
+      unsubscribe();
+    },
+  );
 }
 
 const _channelId = '11111111-1111-4111-8111-111111111111';