From daffef7d045e3b7be5d892a3c8c0c3a7329e30bc Mon Sep 17 00:00:00 2001 From: Tom Brow Date: Thu, 13 Aug 2026 14:24:35 -0700 Subject: [PATCH 1/2] fix(mobile): unwrap observer telemetry batches Signed-off-by: Tom Brow Co-authored-by: Codex Ai-assisted: true --- .../agent_activity/observer_subscription.dart | 31 ++- .../observer_subscription_test.dart | 202 ++++++++++++++++++ 2 files changed, 229 insertions(+), 4 deletions(-) diff --git a/mobile/lib/features/channels/agent_activity/observer_subscription.dart b/mobile/lib/features/channels/agent_activity/observer_subscription.dart index c15686c582..1ee94eb990 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,9 +190,15 @@ 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; + for (final frame in frames) { + _storeFrame(normalizedAgent, frame); + } + } + + void _storeFrame(String normalizedAgent, ObserverFrame frame) { final dedupeKey = '${frame.seq}:${frame.timestamp}'; final dedupeKeys = _dedupeKeysByAgent.putIfAbsent( normalizedAgent, @@ -220,7 +227,7 @@ class ObserverRelayNotifier extends Notifier { _emit(connection: ObserverConnectionState.open); } - ObserverFrame? _decryptFrame( + List? _decryptFrames( NostrEvent event, String normalizedAgent, String privHex, @@ -232,7 +239,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 e7cdb03801..16d82a5c69 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,208 @@ 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('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 { From 9154fc3b7387dd4b0d28f9ff5b094487390cfbb8 Mon Sep 17 00:00:00 2001 From: Tom Brow Date: Fri, 14 Aug 2026 11:26:16 -0700 Subject: [PATCH 2/2] fix(mobile): publish observer batches once Signed-off-by: Tom Brow --- .../agent_activity/observer_subscription.dart | 17 +++-- .../observer_subscription_test.dart | 63 +++++++++++++++++++ 2 files changed, 75 insertions(+), 5 deletions(-) diff --git a/mobile/lib/features/channels/agent_activity/observer_subscription.dart b/mobile/lib/features/channels/agent_activity/observer_subscription.dart index 1ee94eb990..bc469c5b32 100644 --- a/mobile/lib/features/channels/agent_activity/observer_subscription.dart +++ b/mobile/lib/features/channels/agent_activity/observer_subscription.dart @@ -193,19 +193,27 @@ class ObserverRelayNotifier extends Notifier { final frames = _decryptFrames(event, normalizedAgent, privHex); if (frames == null) return; + var storageChanged = false; for (final frame in frames) { - _storeFrame(normalizedAgent, frame); + if (_storeFrame(normalizedAgent, frame)) { + storageChanged = true; + } + } + + if (storageChanged) { + _errorMessage = null; + _emit(connection: ObserverConnectionState.open); } } - void _storeFrame(String normalizedAgent, ObserverFrame frame) { + 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( @@ -223,8 +231,7 @@ class ObserverRelayNotifier extends Notifier { frames.removeRange(0, removeCount); } - _errorMessage = null; - _emit(connection: ObserverConnectionState.open); + return true; } List? _decryptFrames( 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 16d82a5c69..f54b0594ef 100644 --- a/mobile/test/features/channels/agent_activity/observer_subscription_test.dart +++ b/mobile/test/features/channels/agent_activity/observer_subscription_test.dart @@ -357,6 +357,69 @@ void main() { 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();