Skip to content

fix(mobile): unwrap batched observer telemetry #5805

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
brow merged 2 commits into from
Aug 14, 2026
Merged

fix(mobile): unwrap batched observer telemetry #5805

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
daffef7
fix(mobile): unwrap observer telemetry batches
brow Aug 13, 2026
9154fc3
fix(mobile): publish observer batches once
brow Aug 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 27 additions & 4 deletions mobile/lib/features/channels/agent_activity/observer_subscription.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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});
Expand Down Expand Up @@ -189,9 +190,15 @@ 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;

for (final frame in frames) {
_storeFrame(normalizedAgent, frame);

@wesbillman wesbillman Aug 14, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Publish once after storing the whole relay envelope

_storeFrame still calls _emit for every inner frame, so unwrapping a batch of N produces N full state publications. Each publication snapshots every retained agent list (_snapshotFrames), and every channel-scoped provider watching this state filters the list and rebuilds its transcript (observer_subscription.dart:350-365); app.dart:87 also keeps the relay provider live app-wide. A normal publisher batch can therefore turn one relay event into repeated sorting, copying, transcript derivation, and UI rebuilds, with quadratic-looking work as the batch grows. This is precisely the backpressure batching is meant to preserve; desktop commits the full envelope and explicitly tests one external-store notification per batch (observerRelayStore.ts, activeAgentTurnsStore.test.mjs). Make _storeFrame mutate/return whether it changed state, then clear the error and call _emit once after the loop. Add a provider-observer regression asserting one state update for a multi-frame batch.

@square-tomb square-tomb Aug 14, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 Addressed in 9154fc3 by publishing at most one successful observer state update per relay envelope, only when storage changes. Added a provider-listener regression that pins one update for a multi-frame batch and none for an all-duplicate replay.

}
}

void _storeFrame(String normalizedAgent, ObserverFrame frame) {
final dedupeKey = '${frame.seq}:${frame.timestamp}';
final dedupeKeys = _dedupeKeysByAgent.putIfAbsent(
normalizedAgent,
Expand Down Expand Up @@ -220,7 +227,7 @@ class ObserverRelayNotifier extends Notifier<ObserverRelayState> {
_emit(connection: ObserverConnectionState.open);
}

ObserverFrame? _decryptFrame(
List<ObserverFrame>? _decryptFrames(
NostrEvent event,
String normalizedAgent,
String privHex,
Expand All @@ -232,7 +239,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);
Expand Down
202 changes: 202 additions & 0 deletions mobile/test/features/channels/agent_activity/observer_subscription_test.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<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('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 {
Expand Down
Loading