fix(mobile): unwrap batched observer telemetry

mobile/lib/features/channels/agent_activity/observer_subscription.dart

@@ -11,6 +11,7 @@ import 'transcript_builder.dart';
 
 /// Maximum observer events to keep per agent.
 const _maxObserverEvents = 800;
+const _observerBatchKind = 'batch';
 
 /// Key for channel-scoped transcript reads.
 typedef ObserverKey = ({String channelId, String agentPubkey});
@@ -189,16 +190,30 @@ class ObserverRelayNotifier extends Notifier<ObserverRelayState> {
       return;
     }
 
-    final frame = _decryptFrame(event, normalizedAgent, privHex);
-    if (frame == null) return;
+    final frames = _decryptFrames(event, normalizedAgent, privHex);
+    if (frames == null) return;
 
+    var storageChanged = false;
+    for (final frame in frames) {
+      if (_storeFrame(normalizedAgent, frame)) {
+        storageChanged = true;
+      }
+    }
+
+    if (storageChanged) {
+      _errorMessage = null;
+      _emit(connection: ObserverConnectionState.open);
+    }
+  }
+
+  bool _storeFrame(String normalizedAgent, ObserverFrame frame) {
     final dedupeKey = '${frame.seq}:${frame.timestamp}';
     final dedupeKeys = _dedupeKeysByAgent.putIfAbsent(
       normalizedAgent,
       () => <String>{},
     );
     if (!dedupeKeys.add(dedupeKey)) {
-      return;
+      return false;
     }
 
     final frames = _framesByAgent.putIfAbsent(
@@ -216,11 +231,10 @@ class ObserverRelayNotifier extends Notifier<ObserverRelayState> {
       frames.removeRange(0, removeCount);
     }
 
-    _errorMessage = null;
-    _emit(connection: ObserverConnectionState.open);
+    return true;
   }
 
-  ObserverFrame? _decryptFrame(
+  List<ObserverFrame>? _decryptFrames(
     NostrEvent event,
     String normalizedAgent,
     String privHex,
@@ -232,7 +246,23 @@ class ObserverRelayNotifier extends Notifier<ObserverRelayState> {
       );
       final plaintext = nip44Decrypt(conversationKey, event.content);
       final json = jsonDecode(plaintext) as Map<String, dynamic>;
-      return ObserverFrame.fromJson(json);
+      final frame = ObserverFrame.fromJson(json);
+      if (frame.kind != _observerBatchKind) {
+        return [frame];
+      }
+
+      final payload = frame.payload;
+      final events = payload is Map<String, dynamic> ? payload['events'] : null;
+      // Preserve malformed envelopes so publisher defects are not silently
+      // discarded, matching the desktop observer consumer.
+      if (events is! List || events.isEmpty) {
+        return [frame];
+      }
+
+      return [
+        for (final inner in events)
+          ObserverFrame.fromJson(inner as Map<String, dynamic>),
+      ];
     } catch (error) {
       _errorMessage = 'Observer event decrypt failed: $error';
       _emit(connection: ObserverConnectionState.error);

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

@@ -289,6 +289,271 @@ void main() {
       expect(otherChannelState.transcript, isEmpty);
     },
   );
+
+  test('expands batch envelopes through ordering and dedupe', () async {
+    final ownerKeychain = nostr.Keys.generate();
+    final agentKeychain = nostr.Keys.generate();
+    final relaySession = _RecordingRelaySession();
+    final container = ProviderContainer(
+      overrides: [
+        relaySessionProvider.overrideWith(() => relaySession),
+        relayConfigProvider.overrideWith(
+          () => _FakeRelayConfigNotifier(nsec: ownerKeychain.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 laterFrame = _observerFrameJson(
+      seq: 2,
+      channelId: channelId,
+      turnId: 'turn-2',
+    );
+    final earlierFrame = _observerFrameJson(
+      seq: 1,
+      channelId: channelId,
+      turnId: 'turn-1',
+    );
+    relaySession.emit(
+      _observerEvent(
+        ownerKeychain: ownerKeychain,
+        agentKeychain: agentKeychain,
+        payload: {
+          'seq': 2,
+          'timestamp': '2026-04-30T12:00:02.000Z',
+          'kind': 'batch',
+          'channelId': channelId,
+          'turnId': 'turn-2',
+          'payload': {
+            'events': [laterFrame, earlierFrame, earlierFrame],
+          },
+        },
+      ),
+    );
+
+    final relayState = container.read(observerRelayProvider);
+    final frames = relayState.framesByAgent[agentKeychain.public];
+    expect(frames?.map((frame) => frame.seq), [1, 2]);
+
+    final state = container.read(observerSubscriptionProvider(key));
+    expect(state.connection, ObserverConnectionState.open);
+    expect(state.transcript, hasLength(2));
+    expect(state.transcript.map((item) => item.id), [
+      'turn:turn-1',
+      'turn:turn-2',
+    ]);
+
+    final otherChannelState = container.read(
+      observerSubscriptionProvider((
+        channelId: 'other-channel',
+        agentPubkey: agentKeychain.public,
+      )),
+    );
+    expect(otherChannelState.transcript, isEmpty);
+  });
+
+  test('publishes one open-state update for a changed batch', () async {
+    final ownerKeychain = nostr.Keys.generate();
+    final agentKeychain = nostr.Keys.generate();
+    final relaySession = _RecordingRelaySession();
+    final container = ProviderContainer(
+      overrides: [
+        relaySessionProvider.overrideWith(() => relaySession),
+        relayConfigProvider.overrideWith(
+          () => _FakeRelayConfigNotifier(nsec: ownerKeychain.nsec),
+        ),
+      ],
+    );
+    addTearDown(container.dispose);
+
+    container.read(
+      observerSubscriptionProvider((
+        channelId: 'test-channel',
+        agentPubkey: agentKeychain.public,
+      )),
+    );
+    await Future<void>.delayed(Duration.zero);
+
+    var openStateUpdates = 0;
+    final listener = container.listen(observerRelayProvider, (_, next) {
+      if (next.connection == ObserverConnectionState.open) {
+        openStateUpdates += 1;
+      }
+    });
+    addTearDown(listener.close);
+
+    final event = _observerEvent(
+      ownerKeychain: ownerKeychain,
+      agentKeychain: agentKeychain,
+      payload: {
+        'seq': 2,
+        'timestamp': '2026-04-30T12:00:02.000Z',
+        'kind': 'batch',
+        'channelId': 'test-channel',
+        'turnId': 'turn-2',
+        'payload': {
+          'events': [
+            _observerFrameJson(
+              seq: 1,
+              channelId: 'test-channel',
+              turnId: 'turn-1',
+            ),
+            _observerFrameJson(
+              seq: 2,
+              channelId: 'test-channel',
+              turnId: 'turn-2',
+            ),
+          ],
+        },
+      },
+    );
+
+    relaySession.emit(event);
+    expect(openStateUpdates, 1);
+
+    relaySession.emit(event);
+    expect(openStateUpdates, 1);
+  });
+
+  test('keeps malformed batch envelopes as singleton frames', () async {
+    final ownerKeychain = nostr.Keys.generate();
+    final agentKeychain = nostr.Keys.generate();
+    final relaySession = _RecordingRelaySession();
+    final container = ProviderContainer(
+      overrides: [
+        relaySessionProvider.overrideWith(() => relaySession),
+        relayConfigProvider.overrideWith(
+          () => _FakeRelayConfigNotifier(nsec: ownerKeychain.nsec),
+        ),
+      ],
+    );
+    addTearDown(container.dispose);
+
+    container.read(
+      observerSubscriptionProvider((
+        channelId: 'test-channel',
+        agentPubkey: agentKeychain.public,
+      )),
+    );
+    await Future<void>.delayed(Duration.zero);
+
+    relaySession.emit(
+      _observerEvent(
+        ownerKeychain: ownerKeychain,
+        agentKeychain: agentKeychain,
+        payload: {
+          'seq': 3,
+          'timestamp': '2026-04-30T12:00:03.000Z',
+          'kind': 'batch',
+          'channelId': 'test-channel',
+          'payload': <String, dynamic>{},
+        },
+      ),
+    );
+
+    final state = container.read(observerRelayProvider);
+    expect(state.connection, ObserverConnectionState.open);
+    expect(state.errorMessage, isNull);
+    expect(state.framesByAgent[agentKeychain.public], hasLength(1));
+    expect(state.framesByAgent[agentKeychain.public]!.single.kind, 'batch');
+  });
+
+  test(
+    'rejects invalid inner batch frames without partial ingestion',
+    () async {
+      final ownerKeychain = nostr.Keys.generate();
+      final agentKeychain = nostr.Keys.generate();
+      final relaySession = _RecordingRelaySession();
+      final container = ProviderContainer(
+        overrides: [
+          relaySessionProvider.overrideWith(() => relaySession),
+          relayConfigProvider.overrideWith(
+            () => _FakeRelayConfigNotifier(nsec: ownerKeychain.nsec),
+          ),
+        ],
+      );
+      addTearDown(container.dispose);
+
+      container.read(
+        observerSubscriptionProvider((
+          channelId: 'test-channel',
+          agentPubkey: agentKeychain.public,
+        )),
+      );
+      await Future<void>.delayed(Duration.zero);
+
+      relaySession.emit(
+        _observerEvent(
+          ownerKeychain: ownerKeychain,
+          agentKeychain: agentKeychain,
+          payload: {
+            'seq': 2,
+            'timestamp': '2026-04-30T12:00:02.000Z',
+            'kind': 'batch',
+            'channelId': 'test-channel',
+            'payload': {
+              'events': [
+                _observerFrameJson(
+                  seq: 1,
+                  channelId: 'test-channel',
+                  turnId: 'turn-1',
+                ),
+                {'seq': 'invalid'},
+              ],
+            },
+          },
+        ),
+      );
+
+      final state = container.read(observerRelayProvider);
+      expect(state.connection, ObserverConnectionState.error);
+      expect(state.errorMessage, contains('Observer event decrypt failed'));
+      expect(state.framesByAgent[agentKeychain.public], isNull);
+    },
+  );
+}
+
+Map<String, dynamic> _observerFrameJson({
+  required int seq,
+  required String channelId,
+  required String turnId,
+}) => {
+  'seq': seq,
+  'timestamp': '2026-04-30T12:00:0$seq.000Z',
+  'kind': 'turn_started',
+  'channelId': channelId,
+  'turnId': turnId,
+  'payload': {
+    'triggeringEventIds': ['$seq'],
+  },
+};
+
+NostrEvent _observerEvent({
+  required nostr.Keys ownerKeychain,
+  required nostr.Keys agentKeychain,
+  required Map<String, dynamic> payload,
+}) {
+  final conversationKey = getConversationKey(
+    agentKeychain.secret,
+    ownerKeychain.public,
+  );
+  final event = nostr.Event.from(
+    kind: EventKind.agentObserverFrame,
+    content: nip44Encrypt(conversationKey, jsonEncode(payload)),
+    tags: [
+      ['p', ownerKeychain.public],
+      ['agent', agentKeychain.public],
+      ['frame', 'telemetry'],
+    ],
+    secretKey: agentKeychain.secret,
+    verify: false,
+  );
+  return NostrEvent.fromJson(event.toMap());
 }
 
 class _RecordingRelaySession extends RelaySessionNotifier {