From 0bb7c60f824a05ac4d8c8569ee1e74d200069b45 Mon Sep 17 00:00:00 2001 From: Tom Brow <106167956+square-tomb@users.noreply.github.com> Date: Fri, 14 Aug 2026 13:18:28 -0700 Subject: [PATCH] fix(mobile): unwrap batched observer telemetry (#5805) ## Summary Buzz Mobile now expands decrypted ACP observer batch envelopes into their inner telemetry frames before sending them through the existing per-agent dedupe, ordering, cap, and channel-filter pipeline. Singleton observer events keep their existing behavior. Malformed batch envelopes remain visible as outer frames, matching the desktop consumer convention, while invalid inner frames use the existing observer decrypt error path. This restores batched agent progress, tool activity, and incremental transcript updates that Mobile previously ignored. ### Related issue Related to #4917. ### Testing Added tests: - [`observer_subscription_test.dart`](https://github.com/block/buzz/blob/main/mobile/test/features/channels/agent_activity/observer_subscription_test.dart) covers valid batches, singleton behavior, malformed envelopes, and invalid inner frames. Full mobile analysis, formatting, file-size validation, and Flutter tests passed. The repository pre-push gate also passed. --------- Signed-off-by: Tom Brow Co-authored-by: Tom Brow Co-authored-by: Codex --- .../agent_activity/observer_subscription.dart | 44 ++- .../observer_subscription_test.dart | 265 ++++++++++++++++++ 2 files changed, 302 insertions(+), 7 deletions(-) diff --git a/mobile/lib/features/channels/agent_activity/observer_subscription.dart b/mobile/lib/features/channels/agent_activity/observer_subscription.dart index c15686c58..bc469c5b3 100644 --- a/mobile/lib/features/channels/agent_activity/observer_subscription.dart +++ b/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 { 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, () => {}, ); if (!dedupeKeys.add(dedupeKey)) { - return; + return false; } final frames = _framesByAgent.putIfAbsent( @@ -216,11 +231,10 @@ class ObserverRelayNotifier extends Notifier { frames.removeRange(0, removeCount); } - _errorMessage = null; - _emit(connection: ObserverConnectionState.open); + return true; } - ObserverFrame? _decryptFrame( + List? _decryptFrames( NostrEvent event, String normalizedAgent, String privHex, @@ -232,7 +246,23 @@ class ObserverRelayNotifier extends Notifier { ); final plaintext = nip44Decrypt(conversationKey, event.content); final json = jsonDecode(plaintext) as Map; - return ObserverFrame.fromJson(json); + final frame = ObserverFrame.fromJson(json); + if (frame.kind != _observerBatchKind) { + return [frame]; + } + + final payload = frame.payload; + final events = payload is Map ? 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), + ]; } catch (error) { _errorMessage = 'Observer event decrypt failed: $error'; _emit(connection: ObserverConnectionState.error); diff --git a/mobile/test/features/channels/agent_activity/observer_subscription_test.dart b/mobile/test/features/channels/agent_activity/observer_subscription_test.dart index e7cdb0380..f54b0594e 100644 --- a/mobile/test/features/channels/agent_activity/observer_subscription_test.dart +++ b/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.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.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.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': {}, + }, + ), + ); + + 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.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 _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 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 {