From 3c883b381e7c15f3c17c75344e3fccfe92bce02e Mon Sep 17 00:00:00 2001 From: Michael Neale Date: Wed, 13 May 2026 16:28:25 +1000 Subject: [PATCH] feat: real-time suggestion delivery via SSE instead of polling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add subscribe_suggestions() in nostr_share.rs — opens a persistent Nostr relay subscription and returns an mpsc receiver - Server spawns a subscription task when SSE connects for owner sessions, pushes Suggestion events through the existing SSE stream - Add Suggestion variant to MessageEvent enum - Frontend listens via useSessionEvents hook — suggestions appear instantly, no polling needed - One-time fetch on mount catches suggestions from while session was closed - Remove 30s polling interval from SuggestionBanner Signed-off-by: Michael Neale --- crates/goose-server/src/routes/reply.rs | 7 ++ .../goose-server/src/routes/session_events.rs | 85 +++++++++++++++++ crates/goose/src/session/nostr_share.rs | 94 +++++++++++++++++-- ui/desktop/openapi.json | 33 +++++++ ui/desktop/src/api/types.gen.ts | 6 ++ ui/desktop/src/components/BaseChat.tsx | 2 + .../src/components/SuggestionBanner.tsx | 37 +++++--- ui/desktop/src/hooks/useChatStream.ts | 4 +- ui/desktop/src/hooks/useSessionEvents.ts | 32 ++++++- 9 files changed, 276 insertions(+), 24 deletions(-) diff --git a/crates/goose-server/src/routes/reply.rs b/crates/goose-server/src/routes/reply.rs index 09fc5b8b93..53b49579a8 100644 --- a/crates/goose-server/src/routes/reply.rs +++ b/crates/goose-server/src/routes/reply.rs @@ -150,6 +150,13 @@ pub enum MessageEvent { ActiveRequests { request_ids: Vec, }, + /// A suggestion received from a collaborator via Nostr channel. + Suggestion { + text: String, + sender_name: Option, + event_id: String, + timestamp: u64, + }, Ping, } diff --git a/crates/goose-server/src/routes/session_events.rs b/crates/goose-server/src/routes/session_events.rs index dc5d3d5401..3f611a7e70 100644 --- a/crates/goose-server/src/routes/session_events.rs +++ b/crates/goose-server/src/routes/session_events.rs @@ -171,6 +171,91 @@ pub async fn session_events( let stream = ReceiverStream::new(rx); let task_bus = bus.clone(); + // Spawn Nostr suggestion subscription for owner channels + let nostr_tx = tx.clone(); + let nostr_session_id = session_id.clone(); + let nostr_state = state.clone(); + tokio::spawn(async move { + use goose::session::nostr_share; + + let channel = match nostr_state + .session_manager() + .get_nostr_channel(&nostr_session_id) + .await + { + Ok(Some(ch)) if ch.role == goose::session::nostr_channel::ChannelRole::Owner => { + tracing::info!( + "Nostr suggestion subscription starting for session {}", + nostr_session_id + ); + ch + } + Ok(Some(_)) => { + tracing::debug!( + "Session {} has channel but not owner role, skipping subscription", + nostr_session_id + ); + return; + } + Ok(None) => return, + Err(e) => { + tracing::warn!( + "Failed to get channel for session {}: {}", + nostr_session_id, + e + ); + return; + } + }; + + let mut rx = match nostr_share::subscribe_suggestions( + &channel.encryption_key, + &channel.event_id, + channel.relays, + ) + .await + { + Ok(rx) => { + tracing::info!( + "Nostr subscription active for session {}, event {}", + nostr_session_id, + channel.event_id + ); + rx + } + Err(e) => { + tracing::warn!( + "Failed to subscribe to Nostr for session {}: {}", + nostr_session_id, + e + ); + return; + } + }; + + while let Some(suggestion) = rx.recv().await { + tracing::info!( + "Received suggestion for session {} from {:?}", + nostr_session_id, + suggestion.sender_name + ); + let event = MessageEvent::Suggestion { + text: suggestion.text, + sender_name: suggestion.sender_name, + event_id: suggestion.event_id, + timestamp: suggestion.timestamp, + }; + let json_str = serde_json::to_string(&serde_json::to_value(&event).unwrap_or_default()) + .unwrap_or_default(); + let frame = format!("data: {}\n\n", json_str); + if nostr_tx.send(frame).await.is_err() { + tracing::debug!("SSE channel closed for session {}", nostr_session_id); + break; + } + } + tracing::debug!("Nostr subscription ended for session {}", nostr_session_id); + }); + tokio::spawn(async move { let bus = task_bus; diff --git a/crates/goose/src/session/nostr_share.rs b/crates/goose/src/session/nostr_share.rs index 4abe32ca74..b2d9f7153d 100644 --- a/crates/goose/src/session/nostr_share.rs +++ b/crates/goose/src/session/nostr_share.rs @@ -450,13 +450,10 @@ impl NostrSuggestionFetcher for LiveNostrClient { install_rustls_crypto_provider(); let client = Client::default(); for relay in relays { - client - .add_relay(relay) - .await - .with_context(|| format!("Failed to add relay {relay}"))?; + let _ = client.add_relay(relay).await; } - client.try_connect(Duration::from_secs(8)).await; + client.try_connect(Duration::from_secs(3)).await; let mut filter = Filter::new() .kind(Kind::Custom(SUGGESTION_EVENT_KIND)) .event(parent_event_id); @@ -467,16 +464,99 @@ impl NostrSuggestionFetcher for LiveNostrClient { .fetch_events_from( relays.iter().map(String::as_str), filter, - Duration::from_secs(10), + Duration::from_secs(5), ) .await .context("Failed to fetch suggestions from Nostr relays")?; - client.shutdown().await; + let _ = client.shutdown().await; Ok(events.into_iter().collect()) } } +/// Subscribe to real-time suggestions via Nostr relay subscription. +/// Returns a tokio mpsc receiver that yields suggestions as they arrive. +/// The subscription runs until the receiver is dropped. +pub async fn subscribe_suggestions( + encryption_key_hex: &str, + original_event_id: &str, + relays: Vec, +) -> Result> { + let relays = normalize_relays(relays); + if relays.is_empty() { + return Err(anyhow!("At least one Nostr relay is required")); + } + + install_rustls_crypto_provider(); + let client = Client::default(); + for relay in &relays { + let _ = client.add_relay(relay).await; + } + client.try_connect(Duration::from_secs(5)).await; + + let parent_id = EventId::parse(original_event_id)?; + let filter = Filter::new() + .kind(Kind::Custom(SUGGESTION_EVENT_KIND)) + .event(parent_id) + .since(Timestamp::now()); + + client + .subscribe(filter, None) + .await + .map_err(|e| anyhow!("Failed to subscribe: {e}"))?; + + let secret_key = SecretKey::parse(encryption_key_hex)?; + let encryption_keys = Keys::new(secret_key.clone()); + + let (tx, rx) = tokio::sync::mpsc::channel::(32); + + tokio::spawn(async move { + let _ = client + .handle_notifications(|notification| { + let tx = tx.clone(); + let secret_key = secret_key.clone(); + let encryption_keys = encryption_keys.clone(); + async move { + if tx.is_closed() { + return Ok(true); + } + if let nostr_sdk::RelayPoolNotification::Event { event, .. } = notification { + if event.kind == Kind::Custom(SUGGESTION_EVENT_KIND) { + if let Ok(decrypted) = nip44::decrypt( + &secret_key, + &encryption_keys.public_key(), + &event.content, + ) { + let (text, sender_name) = if let Ok(payload) = + serde_json::from_str::(&decrypted) + { + (payload.text, payload.sender_name) + } else { + (decrypted, None) + }; + + let suggestion = Suggestion { + text, + sender_name, + event_id: event.id.to_hex(), + timestamp: event.created_at.as_secs(), + }; + if tx.send(suggestion).await.is_err() { + return Ok(true); + } + } + } + } + Ok(false) + } + }) + .await; + let _ = client.shutdown().await; + }); + + Ok(rx) +} + #[cfg(test)] mod tests { use super::*; diff --git a/ui/desktop/openapi.json b/ui/desktop/openapi.json index b82c545742..913fab2269 100644 --- a/ui/desktop/openapi.json +++ b/ui/desktop/openapi.json @@ -6490,6 +6490,39 @@ } } }, + { + "type": "object", + "description": "A suggestion received from a collaborator via Nostr channel.", + "required": [ + "text", + "event_id", + "timestamp", + "type" + ], + "properties": { + "event_id": { + "type": "string" + }, + "sender_name": { + "type": "string", + "nullable": true + }, + "text": { + "type": "string" + }, + "timestamp": { + "type": "integer", + "format": "int64", + "minimum": 0 + }, + "type": { + "type": "string", + "enum": [ + "Suggestion" + ] + } + } + }, { "type": "object", "required": [ diff --git a/ui/desktop/src/api/types.gen.ts b/ui/desktop/src/api/types.gen.ts index 63c87f51d7..622a8f8ef7 100644 --- a/ui/desktop/src/api/types.gen.ts +++ b/ui/desktop/src/api/types.gen.ts @@ -752,6 +752,12 @@ export type MessageEvent = { } | { request_ids: Array; type: 'ActiveRequests'; +} | { + event_id: string; + sender_name?: string | null; + text: string; + timestamp: number; + type: 'Suggestion'; } | { type: 'Ping'; }; diff --git a/ui/desktop/src/components/BaseChat.tsx b/ui/desktop/src/components/BaseChat.tsx index c6e3532676..3bc6e11195 100644 --- a/ui/desktop/src/components/BaseChat.tsx +++ b/ui/desktop/src/components/BaseChat.tsx @@ -120,6 +120,7 @@ export default function BaseChat({ tokenState, notifications: toolCallNotifications, onMessageUpdate, + setSuggestionHandler, } = useChatStream({ sessionId, onStreamFinish, @@ -495,6 +496,7 @@ export default function BaseChat({ handleSubmit({ msg: `[Suggestion from collaborator]: ${text}`, images: [] })} + setSuggestionHandler={setSuggestionHandler} />
void; + setSuggestionHandler: (handler: ((suggestion: SuggestionEvent) => void) | null) => void; } const SENDER_NAME_KEY = 'goose_suggestion_sender_name'; @@ -34,7 +36,7 @@ function saveSenderName(name: string) { } } -export default function SuggestionBanner({ sessionId, onAcceptSuggestion }: SuggestionBannerProps) { +export default function SuggestionBanner({ sessionId, onAcceptSuggestion, setSuggestionHandler }: SuggestionBannerProps) { const [role, setRole] = useState(null); const [checked, setChecked] = useState(false); const [suggestions, setSuggestions] = useState([]); @@ -63,27 +65,32 @@ export default function SuggestionBanner({ sessionId, onAcceptSuggestion }: Sugg return () => { cancelled = true; }; }, [sessionId]); - // For owners, poll for suggestions on mount and every 30s + // For owners: fetch existing suggestions once, then listen for new ones via SSE useEffect(() => { if (role !== 'owner') return; - const fetchSuggestions = async () => { - setIsLoading(true); - try { - const resp = await getSuggestions({ path: { session_id: sessionId }, throwOnError: false }); + // One-time fetch for suggestions that arrived while session was closed + setIsLoading(true); + getSuggestions({ path: { session_id: sessionId }, throwOnError: false }) + .then((resp) => { if (resp.data) { setSuggestions(resp.data); } - } catch { - } finally { - setIsLoading(false); - } - }; + }) + .catch(() => {}) + .finally(() => setIsLoading(false)); - fetchSuggestions(); - const interval = setInterval(fetchSuggestions, 30000); - return () => clearInterval(interval); - }, [role, sessionId]); + // Listen for real-time suggestions via SSE + setSuggestionHandler((suggestion: SuggestionEvent) => { + setSuggestions((prev) => { + // Deduplicate by eventId + if (prev.some((s) => s.eventId === suggestion.eventId)) return prev; + return [...prev, suggestion]; + }); + }); + + return () => setSuggestionHandler(null); + }, [role, sessionId, setSuggestionHandler]); const handleSend = useCallback(async () => { if (!suggestionText.trim()) return; diff --git a/ui/desktop/src/hooks/useChatStream.ts b/ui/desktop/src/hooks/useChatStream.ts index 8895aff303..7b8310dc0d 100644 --- a/ui/desktop/src/hooks/useChatStream.ts +++ b/ui/desktop/src/hooks/useChatStream.ts @@ -58,6 +58,7 @@ interface UseChatStreamReturn { newContent: string, editType?: 'fork' | 'edit' ) => Promise; + setSuggestionHandler: (handler: ((suggestion: import('./useSessionEvents').SuggestionEvent) => void) | null) => void; } interface StreamState { @@ -364,7 +365,7 @@ export function useChatStream({ const [state, dispatch] = useReducer(streamReducer, initialState); // Long-lived SSE connection for this session - const { addListener, setActiveRequestsHandler } = useSessionEvents(sessionId); + const { addListener, setActiveRequestsHandler, setSuggestionHandler } = useSessionEvents(sessionId); // Track the active request for cancellation (includes the session that started it) const activeRequestIdRef = useRef(null); @@ -1107,5 +1108,6 @@ export function useChatStream({ tokenState: state.tokenState, notifications: notificationsMap, onMessageUpdate, + setSuggestionHandler, }; } diff --git a/ui/desktop/src/hooks/useSessionEvents.ts b/ui/desktop/src/hooks/useSessionEvents.ts index 95c74947cb..dd59fa6e8d 100644 --- a/ui/desktop/src/hooks/useSessionEvents.ts +++ b/ui/desktop/src/hooks/useSessionEvents.ts @@ -14,9 +14,18 @@ export type SessionEvent = MessageEvent & { type EventHandler = (event: SessionEvent) => void; type ActiveRequestsHandler = (requestIds: string[]) => void; +export interface SuggestionEvent { + text: string; + senderName?: string | null; + eventId: string; + timestamp: number; +} +type SuggestionHandler = (suggestion: SuggestionEvent) => void; + export function useSessionEvents(sessionId: string) { const listenersRef = useRef(new Map>()); const activeRequestsHandlerRef = useRef(null); + const suggestionHandlerRef = useRef(null); const abortRef = useRef(null); const [connected, setConnected] = useState(false); @@ -70,6 +79,23 @@ export function useSessionEvents(sessionId: string) { const sessionEvent = event as SessionEvent; const routingId = sessionEvent.chat_request_id ?? sessionEvent.request_id; + // Suggestion events from Nostr channel subscription + if (sessionEvent.type === 'Suggestion') { + const raw = sessionEvent as unknown as { + text: string; + sender_name?: string | null; + event_id: string; + timestamp: number; + }; + suggestionHandlerRef.current?.({ + text: raw.text, + senderName: raw.sender_name, + eventId: raw.event_id, + timestamp: raw.timestamp, + }); + continue; + } + // ActiveRequests events notify the client about in-flight requests // it can reattach to (e.g. after a remount). if (sessionEvent.type === 'ActiveRequests') { @@ -192,5 +218,9 @@ export function useSessionEvents(sessionId: string) { activeRequestsHandlerRef.current = handler; }, []); - return { connected, addListener, setActiveRequestsHandler }; + const setSuggestionHandler = useCallback((handler: SuggestionHandler | null) => { + suggestionHandlerRef.current = handler; + }, []); + + return { connected, addListener, setActiveRequestsHandler, setSuggestionHandler }; }