diff --git a/desktop/src/features/agents/agentWorkingSignal.test.mjs b/desktop/src/features/agents/agentWorkingSignal.test.mjs index 9d9ae31bd6f..26890058346 100644 --- a/desktop/src/features/agents/agentWorkingSignal.test.mjs +++ b/desktop/src/features/agents/agentWorkingSignal.test.mjs @@ -3,6 +3,7 @@ import { beforeEach, describe, it } from "node:test"; import { getAgentWorkingState, + getWorkingAgentPubkeysForConversation, getWorkingAgentPubkeysForChannel, getWorkingChannels, reportChannelBotTyping, @@ -76,6 +77,45 @@ describe("getAgentWorkingState", () => { assert.equal(elsewhere.channels.length, 1); }); + it("keeps thread turns visible through the parent channel scope", () => { + startTurn(AGENT, "chan-1", "thread-turn"); + syncAgentTurnsFromEvents(AGENT, [ + makeEvent({ + channelId: "chan-1", + conversationId: "thread-conversation", + turnId: "thread-turn", + seq: 2, + }), + ]); + + assert.deepEqual(getWorkingAgentPubkeysForChannel("chan-1"), [AGENT]); + assert.equal(getAgentWorkingState(AGENT, "chan-1").working, true); + }); + + it("scopes thread activity by the derived conversation identity", () => { + syncAgentTurnsFromEvents(AGENT, [ + makeEvent({ + channelId: "chan-1", + conversationId: "thread-a", + turnId: "thread-a-turn", + }), + ]); + syncAgentTurnsFromEvents(AGENT_2, [ + makeEvent({ + channelId: "chan-1", + conversationId: "thread-b", + turnId: "thread-b-turn", + }), + ]); + + assert.deepEqual(getWorkingAgentPubkeysForConversation("thread-a"), [ + AGENT, + ]); + assert.deepEqual(getWorkingAgentPubkeysForConversation("thread-b"), [ + AGENT_2, + ]); + }); + it("falls back to typing when no observer turns exist", () => { reportChannelBotTyping("chan-1", [AGENT]); const state = getAgentWorkingState(AGENT, "chan-1"); diff --git a/desktop/src/features/agents/agentWorkingSignal.ts b/desktop/src/features/agents/agentWorkingSignal.ts index d75ea7ccc2d..d607b17ae7f 100644 --- a/desktop/src/features/agents/agentWorkingSignal.ts +++ b/desktop/src/features/agents/agentWorkingSignal.ts @@ -4,6 +4,7 @@ import { normalizePubkey } from "@/shared/lib/pubkey"; import { type ActiveChannelTurnSummary, getActiveTurnsByChannel, + getActiveAgentsForConversation, getActiveTurnsForAgent, subscribeActiveAgentTurns, } from "./activeAgentTurnsStore"; @@ -296,6 +297,18 @@ export function getWorkingAgentPubkeysForChannel( return result; } +export function mergeWorkingAgentPubkeys( + ...pubkeyLists: readonly (readonly string[])[] +): string[] { + const merged = new Set(); + for (const pubkeyList of pubkeyLists) { + for (const pubkey of pubkeyList) { + merged.add(normalizePubkey(pubkey)); + } + } + return merged.size === 0 ? EMPTY_PUBKEYS : [...merged]; +} + // ── Hooks ──────────────────────────────────────────────────────────────────── /** Working state for one agent, optionally scoped to a channel. */ @@ -325,6 +338,27 @@ export function useChannelWorkingAgentPubkeys( ); } +/** Normalized pubkeys of agents working in a conversation. */ +export function useConversationWorkingAgentPubkeys( + conversationId: string | null | undefined, + fallbackPubkeys: readonly string[] = EMPTY_PUBKEYS, +): string[] { + const observerPubkeys = React.useSyncExternalStore( + subscribeAgentWorkingSignal, + () => getWorkingAgentPubkeysForConversation(conversationId), + ); + return React.useMemo( + () => mergeWorkingAgentPubkeys(observerPubkeys, fallbackPubkeys), + [fallbackPubkeys, observerPubkeys], + ); +} + +export function getWorkingAgentPubkeysForConversation( + conversationId: string | null | undefined, +): string[] { + return getActiveAgentsForConversation(conversationId); +} + /** Community-switch reset (see resetCommunityState in useCommunityInit). */ export function resetAgentWorkingSignal() { typingByChannel.clear(); diff --git a/desktop/src/features/agents/conversationId.test.mjs b/desktop/src/features/agents/conversationId.test.mjs new file mode 100644 index 00000000000..7881edb0ada --- /dev/null +++ b/desktop/src/features/agents/conversationId.test.mjs @@ -0,0 +1,44 @@ +import assert from "node:assert/strict"; +import { describe, it } from "node:test"; + +import { + deriveAgentConversationId, + deriveAgentConversationIdOrNull, +} from "./conversationId.ts"; + +describe("deriveAgentConversationId", () => { + it("matches Rust conversation identity vectors", () => { + assert.equal( + deriveAgentConversationId( + "00112233-4455-6677-8899-aabbccddeeff", + "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", + ), + "7415ce56-7adc-d430-f133-c5e06a8e5113", + ); + assert.equal( + deriveAgentConversationId( + "11111111-2222-3333-4444-555555555555", + "abcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcdefabcd", + ), + "026dfba8-bd95-7847-6709-920a0e6d9b97", + ); + }); + + it("returns null for malformed channel or root IDs", () => { + assert.equal( + deriveAgentConversationIdOrNull( + "not-a-uuid", + "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", + ), + null, + ); + assert.equal( + deriveAgentConversationIdOrNull( + "00112233-4455-6677-8899-aabbccddeeff", + "not-an-event-id", + ), + null, + ); + assert.equal(deriveAgentConversationIdOrNull(null, null), null); + }); +}); diff --git a/desktop/src/features/agents/conversationId.ts b/desktop/src/features/agents/conversationId.ts new file mode 100644 index 00000000000..716eb27e6cb --- /dev/null +++ b/desktop/src/features/agents/conversationId.ts @@ -0,0 +1,56 @@ +import { sha256 } from "@noble/hashes/sha2.js"; + +const CONVERSATION_DOMAIN = new TextEncoder().encode( + "buzz-acp-conversation-v1", +); + +function decodeUuid(uuid: string): Uint8Array { + const hex = uuid.replaceAll("-", ""); + if (!/^[0-9a-f]{32}$/i.test(hex)) { + throw new Error(`Invalid UUID: ${uuid}`); + } + return Uint8Array.from({ length: 16 }, (_, index) => + Number.parseInt(hex.slice(index * 2, index * 2 + 2), 16), + ); +} + +function formatUuid(bytes: Uint8Array): string { + const hex = Array.from(bytes, (byte) => byte.toString(16).padStart(2, "0")); + return [ + hex.slice(0, 4).join(""), + hex.slice(4, 6).join(""), + hex.slice(6, 8).join(""), + hex.slice(8, 10).join(""), + hex.slice(10, 16).join(""), + ].join("-"); +} + +export function deriveAgentConversationId( + channelId: string, + rootEventId: string, +): string { + if (!/^[0-9a-f]{64}$/.test(rootEventId)) { + throw new Error(`Invalid root event ID: ${rootEventId}`); + } + const channelBytes = decodeUuid(channelId); + const rootBytes = new TextEncoder().encode(rootEventId); + const input = new Uint8Array( + CONVERSATION_DOMAIN.length + channelBytes.length + rootBytes.length, + ); + input.set(CONVERSATION_DOMAIN); + input.set(channelBytes, CONVERSATION_DOMAIN.length); + input.set(rootBytes, CONVERSATION_DOMAIN.length + channelBytes.length); + return formatUuid(sha256(input).slice(0, 16)); +} + +export function deriveAgentConversationIdOrNull( + channelId: string | null | undefined, + rootEventId: string | null | undefined, +): string | null { + if (!channelId || !rootEventId) return null; + try { + return deriveAgentConversationId(channelId, rootEventId); + } catch { + return null; + } +} diff --git a/desktop/src/features/agents/lib/personaCatalogRelay.test.mjs b/desktop/src/features/agents/lib/personaCatalogRelay.test.mjs index fbaf1f52742..f3727598c01 100644 --- a/desktop/src/features/agents/lib/personaCatalogRelay.test.mjs +++ b/desktop/src/features/agents/lib/personaCatalogRelay.test.mjs @@ -356,7 +356,7 @@ test("test_foreign_entry_with_no_local_copy_stays_unselected", () => { BOB, ); - assert.equal(personas[0].id, "catalog:" + ALICE + ":reviewer"); + assert.equal(personas[0].id, `catalog:${ALICE}:reviewer`); assert.equal(personas[0].isActive, false); }); @@ -377,7 +377,7 @@ test("test_catalog_source_match_is_scoped_to_the_publishing_owner", () => { ALICE, ); - assert.equal(personas[0].id, "catalog:" + BOB + ":reviewer"); + assert.equal(personas[0].id, `catalog:${BOB}:reviewer`); assert.equal(personas[0].isActive, false); }); diff --git a/desktop/src/features/channels/ui/ChannelPane.tsx b/desktop/src/features/channels/ui/ChannelPane.tsx index 92fa172ff5a..a1895764c2e 100644 --- a/desktop/src/features/channels/ui/ChannelPane.tsx +++ b/desktop/src/features/channels/ui/ChannelPane.tsx @@ -38,7 +38,11 @@ import { getThreadPanelLayout } from "@/features/channels/lib/threadPanelLayout" import { useThreadViewMode } from "@/features/channels/lib/threadViewModePreference"; import { useThreadViewModeSwitch } from "@/features/channels/ui/useThreadViewModeSwitch"; import { useFocusDrawerPresence } from "@/features/channels/ui/useFocusDrawerPresence"; -import { useChannelWorkingAgentPubkeys } from "@/features/agents/agentWorkingSignal"; +import { + useChannelWorkingAgentPubkeys, + useConversationWorkingAgentPubkeys, +} from "@/features/agents/agentWorkingSignal"; +import { deriveAgentConversationIdOrNull } from "@/features/agents/conversationId"; import { BotActivityComposerAction } from "@/features/channels/ui/BotActivityBar"; import { ChannelComposerActivityAccessory } from "@/features/channels/ui/ChannelComposerActivityAccessory"; import { @@ -423,8 +427,16 @@ export const ChannelPane = React.memo(function ChannelPane({ ) === index, ); }, [botTypingEntries, openThreadHeadId]); + const threadComposerConversationId = React.useMemo( + () => deriveAgentConversationIdOrNull(activeChannel?.id, openThreadHeadId), + [activeChannel?.id, openThreadHeadId], + ); + const threadComposerWorkingBotPubkeys = useConversationWorkingAgentPubkeys( + threadComposerConversationId, + threadComposerBotTypingPubkeys, + ); const hasThreadComposerBotActivity = - threadComposerBotTypingPubkeys.length > 0; + threadComposerWorkingBotPubkeys.length > 0; const directMessageIntro = React.useMemo( () => buildDirectMessageIntro({ @@ -894,7 +906,7 @@ export const ChannelPane = React.memo(function ChannelPane({ onOpenAgentSession={onOpenAgentSession} openAgentSessionPubkey={openAgentSessionPubkey} profiles={profiles} - workingBotPubkeys={threadComposerBotTypingPubkeys} + workingBotPubkeys={threadComposerWorkingBotPubkeys} variant="inline" /> ) : null diff --git a/mobile/lib/shared/relay/relay_session.dart b/mobile/lib/shared/relay/relay_session.dart index 9c167e10b16..b7b5960eb02 100644 --- a/mobile/lib/shared/relay/relay_session.dart +++ b/mobile/lib/shared/relay/relay_session.dart @@ -75,6 +75,8 @@ typedef RelaySocketFactory = }); class RelaySessionNotifier extends Notifier { + static const _shortBackgroundThreshold = Duration(seconds: 5); + RelaySessionNotifier({ http.Client? httpClient, RelaySocketFactory socketFactory = RelaySocket.new, @@ -103,6 +105,7 @@ class RelaySessionNotifier extends Notifier { int _subIdCounter = 0; bool _disposed = false; bool _paused = false; + DateTime? _pausedAt; bool _hasConnectedOnce = false; int _connectionGeneration = 0; @@ -307,6 +310,10 @@ class RelaySessionNotifier extends Notifier { /// Force a reconnect (e.g., returning from background). Future reconnect() async { + // Invalidate callbacks from the socket being replaced before closing it. + // Some WebSocket implementations deliver onDone asynchronously, which + // must not schedule a second reconnect while this one is in progress. + _connectionGeneration++; await _socket?.disconnect(); _reconnectDelayMs = _baseReconnectDelayMs; final config = ref.read(relayConfigProvider); @@ -315,6 +322,7 @@ class RelaySessionNotifier extends Notifier { /// Called by the app lifecycle provider when the app goes to background. void onAppPaused() { + _pausedAt = DateTime.now(); _backgroundGraceTimer?.cancel(); _backgroundGraceTimer = Timer(const Duration(seconds: 5), _pauseNow); } @@ -330,12 +338,30 @@ class RelaySessionNotifier extends Notifier { /// Called by the app lifecycle provider when the app returns to foreground. void onAppResumed() { + final pausedAt = _pausedAt; + _pausedAt = null; _paused = false; _backgroundGraceTimer?.cancel(); _backgroundGraceTimer = null; - // If still connected, nothing to do — the socket survived the background - // grace window. + // A suspended isolate may not run the grace timer. Preserve a very short + // app switch only when the connected socket saw a recent data frame; + // otherwise replace it so a half-open socket cannot remain "connected". + if (pausedAt != null) { + final now = DateTime.now(); + final socket = _socket; + final hasRecentInbound = + socket?.state == SocketState.connected && + socket?.lastInboundAt != null && + now.difference(socket!.lastInboundAt!) <= _shortBackgroundThreshold; + if (now.difference(pausedAt) < _shortBackgroundThreshold && + hasRecentInbound) { + return; + } + unawaited(reconnect()); + return; + } + if (state.status == SessionStatus.connected) return; // Cancel any in-flight reconnect backoff timer so we reconnect immediately @@ -639,6 +665,7 @@ class RelaySessionNotifier extends Notifier { _reconnectTimer?.cancel(); _flushTimer?.cancel(); _backgroundGraceTimer?.cancel(); + _pausedAt = null; _cancelAllHistory(null); _rejectAllPending(null); _recentDeliveryKeys.clear(); diff --git a/mobile/lib/shared/relay/relay_socket.dart b/mobile/lib/shared/relay/relay_socket.dart index 267b030391d..adab905060e 100644 --- a/mobile/lib/shared/relay/relay_socket.dart +++ b/mobile/lib/shared/relay/relay_socket.dart @@ -3,6 +3,7 @@ import 'dart:convert'; import 'package:flutter/foundation.dart'; import 'package:nostr/nostr.dart' as nostr; +import 'package:web_socket_channel/io.dart'; import 'package:web_socket_channel/web_socket_channel.dart'; import 'nostr_models.dart'; @@ -30,6 +31,8 @@ Exception classifyRelayAuthFailure(String message) { } class RelaySocket { + static const _pingInterval = Duration(seconds: 30); + final String _wsUrl; final String? _nsec; final void Function(List message) _onMessage; @@ -41,9 +44,11 @@ class RelaySocket { SocketState _state = SocketState.disconnected; Completer? _authCompleter; Timer? _authTimeout; + DateTime? _lastInboundAt; String? _pendingAuthEventId; SocketState get state => _state; + DateTime? get lastInboundAt => _lastInboundAt; RelaySocket({ required String wsUrl, @@ -63,7 +68,14 @@ class RelaySocket { _state = SocketState.connecting; try { - _channel = WebSocketChannel.connect(Uri.parse(_wsUrl)); + // Use dart:io's protocol-level ping watchdog. The relay sends a Ping + // every 30 seconds, but those control frames are handled below the + // channel stream; an inbound-data-only watchdog would falsely kill a + // healthy idle subscription after 60 seconds. + _channel = IOWebSocketChannel.connect( + Uri.parse(_wsUrl), + pingInterval: _pingInterval, + ); await _channel!.ready; } catch (e) { _state = SocketState.disconnected; @@ -143,6 +155,7 @@ class RelaySocket { _subscription = null; _authTimeout?.cancel(); _authTimeout = null; + _lastInboundAt = null; _pendingAuthEventId = null; } @@ -153,6 +166,7 @@ class RelaySocket { } void _handleRawMessage(dynamic raw) { + _lastInboundAt = DateTime.now(); final String text; if (raw is String) { text = raw; diff --git a/mobile/test/shared/relay/relay_session_test.dart b/mobile/test/shared/relay/relay_session_test.dart index 6cd601f56b0..78d48f800fc 100644 --- a/mobile/test/shared/relay/relay_session_test.dart +++ b/mobile/test/shared/relay/relay_session_test.dart @@ -260,6 +260,84 @@ void main() { expect(session.state.status, SessionStatus.disconnected); }); + test('reconnects and replays live subscriptions after resume', () async { + final sockets = <_ControlledRelaySocket>[]; + final keychain = nostr.Keys.generate(); + final session = RelaySessionNotifier( + socketFactory: + ({ + required wsUrl, + required nsec, + required onMessage, + required onConnected, + required onDisconnected, + }) { + final socket = _ControlledRelaySocket( + wsUrl: wsUrl, + nsec: nsec, + onMessage: onMessage, + onConnected: onConnected, + onDisconnected: onDisconnected, + ); + sockets.add(socket); + return socket; + }, + ); + final container = ProviderContainer( + overrides: [ + relaySessionProvider.overrideWith(() => session), + relayConfigProvider.overrideWith( + () => _FakeRelayConfigNotifier( + baseUrl: 'https://relay.example', + nsec: keychain.nsec, + ), + ), + authProvider.overrideWith(() => _AuthenticatedAuthNotifier()), + ], + ); + addTearDown(container.dispose); + final providerSubscription = container.listen( + relaySessionProvider, + (_, _) {}, + ); + addTearDown(providerSubscription.close); + await container.read(authProvider.future); + await Future.delayed(Duration.zero); + + sockets.single.connectSuccessfully(); + final unsubscribe = await session.subscribe( + const NostrFilter(kinds: [1], limit: 1), + (_) {}, + ); + addTearDown(unsubscribe); + sockets.single.emitEose('l-1'); + + sockets.single.inboundAt = DateTime.now(); + session.onAppPaused(); + session.onAppResumed(); + await Future.delayed(Duration.zero); + expect(sockets, hasLength(1)); + + sockets.single.inboundAt = DateTime.now().subtract( + const Duration(seconds: 10), + ); + session.onAppPaused(); + session.onAppResumed(); + await Future.delayed(Duration.zero); + expect(sockets, hasLength(2)); + sockets.last.connectSuccessfully(); + expect( + sockets.last.sent, + contains( + predicate>((payload) { + return payload.length >= 3 && + payload.first == 'REQ' && + payload[1] == 'l-1'; + }), + ), + ); + }); + test('delivers the same live event to each matching subscription', () async { final session = RelaySessionNotifier(); final firstEvents = []; @@ -372,6 +450,10 @@ class _AuthenticatedAuthNotifier extends AuthNotifier { class _ControlledRelaySocket extends RelaySocket { final void Function() _connected; final void Function(Object? error) _disconnected; + final void Function(List) _message; + final List> sent = []; + bool _isConnected = false; + DateTime? inboundAt; _ControlledRelaySocket({ required super.wsUrl, @@ -380,17 +462,38 @@ class _ControlledRelaySocket extends RelaySocket { required super.onConnected, required super.onDisconnected, }) : _connected = onConnected, - _disconnected = onDisconnected; + _disconnected = onDisconnected, + _message = onMessage; @override Future connect() async {} @override - void dispose() {} + SocketState get state => + _isConnected ? SocketState.connected : SocketState.disconnected; - void connectSuccessfully() => _connected(); + @override + Future disconnect() async { + _isConnected = false; + } + + @override + void dispose() => _isConnected = false; + + @override + DateTime? get lastInboundAt => inboundAt; + + @override + void send(List payload) => sent.add(payload); + + void connectSuccessfully() { + _isConnected = true; + _connected(); + } void disconnectWith(Object? error) => _disconnected(error); + + void emitEose(String subId) => _message(['EOSE', subId]); } const _channelId = '11111111-1111-4111-8111-111111111111';