feat: real-time suggestion delivery via SSE instead of polling

- 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 <michael.neale@gmail.com>

crates/goose-server/src/routes/reply.rs

@@ -150,6 +150,13 @@ pub enum MessageEvent {
     ActiveRequests {
         request_ids: Vec<String>,
     },
+    /// A suggestion received from a collaborator via Nostr channel.
+    Suggestion {
+        text: String,
+        sender_name: Option<String>,
+        event_id: String,
+        timestamp: u64,
+    },
     Ping,
 }
 

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;
 

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<String>,
+) -> Result<tokio::sync::mpsc::Receiver<Suggestion>> {
+    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::<Suggestion>(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::<SuggestionPayload>(&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::*;

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": [

ui/desktop/src/api/types.gen.ts

@@ -752,6 +752,12 @@ export type MessageEvent = {
 } | {
     request_ids: Array<string>;
     type: 'ActiveRequests';
+} | {
+    event_id: string;
+    sender_name?: string | null;
+    text: string;
+    timestamp: number;
+    type: 'Suggestion';
 } | {
     type: 'Ping';
 };

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({
         <SuggestionBanner
           sessionId={sessionId}
           onAcceptSuggestion={(text) => handleSubmit({ msg: `[Suggestion from collaborator]: ${text}`, images: [] })}
+          setSuggestionHandler={setSuggestionHandler}
         />
 
         <div

ui/desktop/src/components/SuggestionBanner.tsx

@@ -3,6 +3,7 @@ import { MessageSquarePlus, Inbox, LoaderCircle, X, Check, Share2, Copy } from '
 import { Button } from './ui/button';
 import { getChannelInfo, getSuggestions, sendSuggestion, shareSessionNostr } from '../api/sdk.gen';
 import { toast } from 'react-toastify';
+import type { SuggestionEvent } from '../hooks/useSessionEvents';
 
 interface Suggestion {
   text: string;
@@ -14,6 +15,7 @@ interface Suggestion {
 interface SuggestionBannerProps {
   sessionId: string;
   onAcceptSuggestion: (text: string) => 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<string | null>(null);
   const [checked, setChecked] = useState(false);
   const [suggestions, setSuggestions] = useState<Suggestion[]>([]);
@@ -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;

ui/desktop/src/hooks/useChatStream.ts

@@ -58,6 +58,7 @@ interface UseChatStreamReturn {
     newContent: string,
     editType?: 'fork' | 'edit'
   ) => Promise<void>;
+  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<string | null>(null);
@@ -1107,5 +1108,6 @@ export function useChatStream({
     tokenState: state.tokenState,
     notifications: notificationsMap,
     onMessageUpdate,
+    setSuggestionHandler,
   };
 }

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<string, Set<EventHandler>>());
   const activeRequestsHandlerRef = useRef<ActiveRequestsHandler | null>(null);
+  const suggestionHandlerRef = useRef<SuggestionHandler | null>(null);
   const abortRef = useRef<AbortController | null>(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 };
 }