mirror of
https://github.com/block/buzz.git
synced 2026-08-18 06:50:31 +02:00
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 <tomb@block.xyz> Co-authored-by: Tom Brow <tomb@block.xyz> Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Tom Brow
co-authored by
Tom Brow
Codex
Tom Brow
Codex
parent
757779bb1e
commit
0bb7c60f82
@@ -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);
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user