cursor presence and real-time doc sync

Cargo.lock

···
  "http",
  "iroh",
  "iroh-gossip",
- "iroh-quinn",
  "iroh-tickets",
  "jacquard",
  "jacquard-common",

crates/weaver-app/Cargo.toml

···
 chrono = { version = "0.4", features = ["wasmbind"] }
 wasm-bindgen = "0.2"
 wasm-bindgen-futures = "0.4"
-web-sys = { version = "0.3", features = ["ServiceWorkerContainer", "ServiceWorker", "ServiceWorkerRegistration", "RegistrationOptions", "Window", "Navigator", "MessageEvent", "console", "Document", "Element", "HtmlImageElement", "Selection", "Range", "Node", "HtmlElement", "TreeWalker", "NodeFilter", "DomTokenList", "Clipboard", "ClipboardItem", "Blob", "BlobPropertyBag", "EventTarget", "InputEvent", "AddEventListenerOptions"] }
+web-sys = { version = "0.3", features = ["ServiceWorkerContainer", "ServiceWorker", "ServiceWorkerRegistration", "RegistrationOptions", "Window", "Navigator", "MessageEvent", "console", "Document", "Element", "HtmlImageElement", "Selection", "Range", "Node", "HtmlElement", "TreeWalker", "NodeFilter", "DomTokenList", "Clipboard", "ClipboardItem", "Blob", "BlobPropertyBag", "EventTarget", "InputEvent", "AddEventListenerOptions", "DomRect", "DomRectList"] }
 js-sys = "0.3"
 gloo-storage = "0.3"
 gloo-timers = "0.3"

crates/weaver-app/assets/styling/editor.css

···
     pointer-events: none;
 }
 
+.remote-selection {
+    position: absolute;
+    pointer-events: none;
+    border-radius: 2px;
+}
+
 .remote-cursor-caret {
     width: 2px;
     height: var(--cursor-height, 18px);

crates/weaver-app/src/collab_context.rs

···
 use std::sync::Arc;
 use weaver_common::transport::CollabNode;
 
+/// Debug state for the collab session, displayed in editor debug panel.
+#[derive(Clone, Default)]
+pub struct CollabDebugState {
+    /// Our node ID
+    pub node_id: Option<String>,
+    /// Our relay URL
+    pub relay_url: Option<String>,
+    /// URI of our published session record
+    pub session_record_uri: Option<String>,
+    /// Number of discovered peers
+    pub discovered_peers: usize,
+    /// Number of connected peers
+    pub connected_peers: usize,
+    /// Whether we've joined the gossip swarm
+    pub is_joined: bool,
+    /// Last error message
+    pub last_error: Option<String>,
+}
+
 /// Context state for the collaboration node.
 ///
 /// This is provided as a Dioxus context and can be accessed by editor components
···
 #[component]
 pub fn CollabProvider(children: Element) -> Element {
     let mut collab_ctx = use_signal(CollabContext::default);
+    let debug_state = use_signal(CollabDebugState::default);
 
     // Spawn the CollabNode on mount
     let _spawn_result = use_resource(move || async move {
···
         }
     });
 
-    // Provide the context
+    // Provide the contexts
     use_context_provider(|| collab_ctx);
+    use_context_provider(|| debug_state);
 
     rsx! { {children} }
 }
···
 pub fn CollabProvider(children: Element) -> Element {
     // On server/native, provide an empty context (collab happens in browser)
     let collab_ctx = use_signal(CollabContext::default);
+    let debug_state = use_signal(CollabDebugState::default);
     use_context_provider(|| collab_ctx);
+    use_context_provider(|| debug_state);
     rsx! { {children} }
 }
 
···
     let ctx = use_context::<Signal<CollabContext>>();
     ctx.read().node.is_some()
 }
+
+/// Hook to get the collab debug state signal.
+/// Returns None if called outside CollabProvider.
+pub fn try_use_collab_debug() -> Option<Signal<CollabDebugState>> {
+    try_use_context::<Signal<CollabDebugState>>()
+}

crates/weaver-app/src/components/editor/component.rs

···
     let mut doc_for_dom = document.clone();
     #[cfg(all(target_arch = "wasm32", target_os = "unknown"))]
     use_effect(move || {
-        tracing::debug!("DOM update effect triggered");
-
-        tracing::debug!(
-            composition_active = doc_for_dom.composition.read().is_some(),
-            cursor = doc_for_dom.cursor.read().offset,
-            "DOM update: checking state"
-        );
+        // tracing::debug!(
+        //     composition_active = doc_for_dom.composition.read().is_some(),
+        //     cursor = doc_for_dom.cursor.read().offset,
+        //     "DOM update: checking state"
+        // );
 
         // Skip DOM updates during IME composition - browser controls the preview
         if doc_for_dom.composition.read().is_some() {
···
             return;
         }
 
-        tracing::debug!(
+        tracing::trace!(
             cursor = doc_for_dom.cursor.read().offset,
             len = doc_for_dom.len_chars(),
             "DOM update proceeding (not in composition)"
···
                     onselect: {
                         let mut doc = document.clone();
                         move |_evt| {
-                            tracing::debug!("onselect fired");
+                            tracing::trace!("onselect fired");
                             let paras = cached_paragraphs();
                             sync_cursor_from_dom(&mut doc, editor_id, &paras);
                             let spans = syntax_spans();
···
                     onselectstart: {
                         let mut doc = document.clone();
                         move |_evt| {
-                            tracing::debug!("onselectstart fired");
+                            tracing::trace!("onselectstart fired");
                             let paras = cached_paragraphs();
                             sync_cursor_from_dom(&mut doc, editor_id, &paras);
                             let spans = syntax_spans();
···
                     onselectionchange: {
                         let mut doc = document.clone();
                         move |_evt| {
-                            tracing::debug!("onselectionchange fired");
+                            tracing::trace!("onselectionchange fired");
                             let paras = cached_paragraphs();
                             sync_cursor_from_dom(&mut doc, editor_id, &paras);
                             let spans = syntax_spans();
···
                     onclick: {
                         let mut doc = document.clone();
                         move |evt| {
-                            tracing::debug!("onclick fired");
+                            tracing::trace!("onclick fired");
                             let paras = cached_paragraphs();
 
                             // Check if click target is a math-clickable element
···
                         let mut doc = document.clone();
                         move |evt: CompositionEvent| {
                             let data = evt.data().data();
-                            tracing::debug!(
+                            tracing::trace!(
                                 data = %data,
                                 "compositionstart"
                             );
···
                             if let Some(sel) = sel {
                                 let (start, end) =
                                     (sel.anchor.min(sel.head), sel.anchor.max(sel.head));
-                                tracing::debug!(
+                                tracing::trace!(
                                     start,
                                     end,
                                     "compositionstart: deleting selection"
···
                             }
 
                             let cursor_offset = doc.cursor.read().offset;
-                            tracing::debug!(
+                            tracing::trace!(
                                 cursor = cursor_offset,
                                 "compositionstart: setting composition state"
                             );
···
                         let mut doc = document.clone();
                         move |evt: CompositionEvent| {
                             let data = evt.data().data();
-                            tracing::debug!(
+                            tracing::trace!(
                                 data = %data,
                                 "compositionupdate"
                             );
···
                         let mut doc = document.clone();
                         move |evt: CompositionEvent| {
                             let final_text = evt.data().data();
-                            tracing::debug!(
+                            tracing::trace!(
                                 data = %final_text,
                                 "compositionend"
                             );
···
                     }
                     div { class: "editor-debug",
                         div { "Cursor: {document.cursor.read().offset}, Chars: {document.len_chars()}" },
+                        // Collab debug info
+                        {
+                            if let Some(debug_state) = crate::collab_context::try_use_collab_debug() {
+                                let ds = debug_state.read();
+                                rsx! {
+                                    div { class: "collab-debug",
+                                        if let Some(ref node_id) = ds.node_id {
+                                            span { title: "{node_id}", "Node: {&node_id[..8.min(node_id.len())]}…" }
+                                        }
+                                        if ds.is_joined {
+                                            span { class: "joined", "✓ Joined" }
+                                        }
+                                        span { "Peers: {ds.discovered_peers}" }
+                                        if let Some(ref err) = ds.last_error {
+                                            span { class: "error", title: "{err}", "⚠" }
+                                        }
+                                    }
+                                }
+                            } else {
+                                rsx! {}
+                            }
+                        },
                         ReportButton {
                             email: "[e-mail]".to_string(),
                             editor_id: "markdown-editor".to_string(),
···
     render_cache: Signal<render::RenderCache>,
 ) -> Element {
     let presence_read = presence.read();
+    let cursor_count = presence_read.len();
     let cursors: Vec<_> = presence_read
         .cursors()
         .map(|(c, cur)| (c.display_name.clone(), c.color, cur.position, cur.selection))
         .collect();
+
+    if cursor_count > 0 {
+        tracing::debug!(
+            "RemoteCursors: {} collaborators, {} with cursors",
+            cursor_count,
+            cursors.len()
+        );
+    }
 
     if cursors.is_empty() {
         return rsx! {};
···
     color: u32,
     offset_map: Vec<super::offset_map::OffsetMapping>,
 ) -> Element {
-    use super::cursor::{get_cursor_rect_relative, CursorRect};
+    use super::cursor::{get_cursor_rect_relative, get_selection_rects_relative};
 
-    // Convert RGBA u32 to CSS color
+    // Convert RGBA u32 to CSS color (fully opaque for cursor)
     let r = (color >> 24) & 0xFF;
     let g = (color >> 16) & 0xFF;
     let b = (color >> 8) & 0xFF;
     let a = (color & 0xFF) as f32 / 255.0;
     let color_css = format!("rgba({}, {}, {}, {})", r, g, b, a);
+    // Semi-transparent version for selection highlight
+    let selection_color_css = format!("rgba({}, {}, {}, 0.25)", r, g, b);
 
     // Get cursor position relative to editor
     let rect = get_cursor_rect_relative(position, &offset_map, "markdown-editor");
+
+    // Get selection rectangles if there's a selection
+    let selection_rects = if let Some((start, end)) = selection {
+        let (start, end) = if start <= end { (start, end) } else { (end, start) };
+        get_selection_rects_relative(start, end, &offset_map, "markdown-editor")
+    } else {
+        vec![]
+    };
 
     let Some(rect) = rect else {
+        tracing::debug!(
+            "RemoteCursorIndicator: no rect for position {} (offset_map len: {})",
+            position,
+            offset_map.len()
+        );
         return rsx! {};
     };
 
+    tracing::trace!(
+        "RemoteCursorIndicator: {} at ({}, {}) h={}, selection_rects={}",
+        display_name,
+        rect.x,
+        rect.y,
+        rect.height,
+        selection_rects.len()
+    );
+
     let style = format!(
         "left: {}px; top: {}px; --cursor-height: {}px; --cursor-color: {};",
         rect.x, rect.y, rect.height, color_css
     );
 
     rsx! {
+        // Selection highlight rectangles (rendered behind cursor)
+        for (i, sel_rect) in selection_rects.iter().enumerate() {
+            div {
+                key: "sel-{i}",
+                class: "remote-selection",
+                style: "left: {sel_rect.x}px; top: {sel_rect.y}px; width: {sel_rect.width}px; height: {sel_rect.height}px; background-color: {selection_color_css};",
+            }
+        }
+
         div {
             class: "remote-cursor",
             style: "{style}",

crates/weaver-app/src/components/editor/cursor.rs

···
 ) -> Option<CursorRect> {
     None
 }
+
+/// A rectangle for part of a selection (one per line).
+#[derive(Debug, Clone, Copy)]
+pub struct SelectionRect {
+    pub x: f64,
+    pub y: f64,
+    pub width: f64,
+    pub height: f64,
+}
+
+/// Get screen rectangles for a selection range, relative to editor.
+///
+/// Returns multiple rects if selection spans multiple lines.
+#[cfg(all(target_family = "wasm", target_os = "unknown"))]
+pub fn get_selection_rects_relative(
+    start: usize,
+    end: usize,
+    offset_map: &[OffsetMapping],
+    editor_id: &str,
+) -> Vec<SelectionRect> {
+    use wasm_bindgen::JsCast;
+
+    if offset_map.is_empty() || start >= end {
+        return vec![];
+    }
+
+    let Some(window) = web_sys::window() else {
+        return vec![];
+    };
+    let Some(document) = window.document() else {
+        return vec![];
+    };
+    let Some(editor) = document.get_element_by_id(editor_id) else {
+        return vec![];
+    };
+    let editor_rect = editor.get_bounding_client_rect();
+
+    // Find mappings for start and end
+    let Some((start_mapping, _)) = find_mapping_for_char(offset_map, start) else {
+        return vec![];
+    };
+    let Some((end_mapping, _)) = find_mapping_for_char(offset_map, end) else {
+        return vec![];
+    };
+
+    // Get containers
+    let start_container = document
+        .get_element_by_id(&start_mapping.node_id)
+        .or_else(|| {
+            let selector = format!("[data-node-id='{}']", start_mapping.node_id);
+            document.query_selector(&selector).ok().flatten()
+        });
+    let end_container = document
+        .get_element_by_id(&end_mapping.node_id)
+        .or_else(|| {
+            let selector = format!("[data-node-id='{}']", end_mapping.node_id);
+            document.query_selector(&selector).ok().flatten()
+        });
+
+    let (Some(start_container), Some(end_container)) = (start_container, end_container) else {
+        return vec![];
+    };
+
+    // Create range
+    let Ok(range) = document.create_range() else {
+        return vec![];
+    };
+
+    // Set start
+    if let Some(child_index) = start_mapping.child_index {
+        let _ = range.set_start(&start_container, child_index as u32);
+    } else if let Ok(container_element) = start_container.clone().dyn_into::<web_sys::HtmlElement>() {
+        let offset_in_range = start - start_mapping.char_range.start;
+        let target_utf16_offset = start_mapping.char_offset_in_node + offset_in_range;
+        if let Ok((text_node, node_offset)) = find_text_node_at_offset(&container_element, target_utf16_offset) {
+            let _ = range.set_start(&text_node, node_offset as u32);
+        }
+    }
+
+    // Set end
+    if let Some(child_index) = end_mapping.child_index {
+        let _ = range.set_end(&end_container, child_index as u32);
+    } else if let Ok(container_element) = end_container.dyn_into::<web_sys::HtmlElement>() {
+        let offset_in_range = end - end_mapping.char_range.start;
+        let target_utf16_offset = end_mapping.char_offset_in_node + offset_in_range;
+        if let Ok((text_node, node_offset)) = find_text_node_at_offset(&container_element, target_utf16_offset) {
+            let _ = range.set_end(&text_node, node_offset as u32);
+        }
+    }
+
+    // Get all rects (one per line)
+    let Some(rects) = range.get_client_rects() else {
+        return vec![];
+    };
+    let mut result = Vec::new();
+
+    for i in 0..rects.length() {
+        if let Some(rect) = rects.get(i) {
+            let rect: web_sys::DomRect = rect;
+            result.push(SelectionRect {
+                x: rect.x() - editor_rect.x(),
+                y: rect.y() - editor_rect.y(),
+                width: rect.width(),
+                height: rect.height().max(16.0),
+            });
+        }
+    }
+
+    result
+}
+
+#[cfg(not(all(target_family = "wasm", target_os = "unknown")))]
+pub fn get_selection_rects_relative(
+    _start: usize,
+    _end: usize,
+    _offset_map: &[OffsetMapping],
+    _editor_id: &str,
+) -> Vec<SelectionRect> {
+    vec![]
+}

crates/weaver-app/src/components/editor/sync.rs

···
     if let Ok(records_container) =
         embeds_map.get_or_create_container("records", loro::LoroList::new())
     {
-        tracing::debug!("Loro embeds map records len: {}", records_container.len());
-
         for i in 0..records_container.len() {
             let Some(value) = records_container.get(i) else {
                 continue;
···
                 .await
             {
                 Ok(html) => {
-                    tracing::debug!("Pre-fetched embed from Loro map: {}", key_uri);
                     resolved.add_embed(key_uri, html, None);
                 }
                 Err(e) => {
···
 
     // Strategy 2: If no embeds found in Loro map, parse markdown text
     if resolved.embed_content.is_empty() {
-        use weaver_common::{collect_refs_from_markdown, ExtractedRef};
+        use weaver_common::{ExtractedRef, collect_refs_from_markdown};
 
         let text = doc.get_text("content");
         let markdown = text.to_string();
 
         if !markdown.is_empty() {
-            tracing::debug!("Falling back to markdown parsing for embeds");
             let refs = collect_refs_from_markdown(&markdown);
 
             for extracted in refs {
···
 
                     match weaver_renderer::atproto::fetch_and_render(&key_uri, fetcher).await {
                         Ok(html) => {
-                            tracing::debug!("Pre-fetched embed from markdown: {}", uri);
                             resolved.add_embed(key_uri, html, None);
                         }
                         Err(e) => {
···
     // Use inline for small diffs, blob for larger ones
     let (blob_ref, inline_diff): (Option<jacquard::types::blob::BlobRef<'static>>, _) =
         if updates.len() <= INLINE_THRESHOLD {
-            tracing::debug!("Using inline diff ({} bytes)", updates.len());
             (None, Some(jacquard::bytes::Bytes::from(updates)))
         } else {
-            tracing::debug!("Using blob diff ({} bytes)", updates.len());
             let mime_type = MimeType::new_static("application/octet-stream");
             let blob = client.upload_blob(updates, mime_type).await.map_err(|e| {
                 WeaverError::InvalidNotebook(format!("Failed to upload diff: {}", e))
···
 
         // Get the last seen diff rkey for this root (if any)
         let after_rkey = root_uri.as_ref().and_then(|uri| {
-            last_seen_diffs.get(uri).and_then(|diff_uri| {
-                diff_uri.rkey().map(|rk| rk.0.to_string())
-            })
+            last_seen_diffs
+                .get(uri)
+                .and_then(|diff_uri| diff_uri.rkey().map(|rk| rk.0.to_string()))
         });
 
         // Load state from this root (skipping already-seen diffs)
-        if let Some(pds_state) = load_edit_state_from_root_id(fetcher, root_id, after_rkey.as_deref()).await? {
+        if let Some(pds_state) =
+            load_edit_state_from_root_id(fetcher, root_id, after_rkey.as_deref()).await?
+        {
             // Import root snapshot into merged doc
             if let Err(e) = merged_doc.import(&pds_state.root_snapshot) {
                 tracing::warn!("Failed to import root snapshot from {}: {:?}", root_did, e);
···
         // Skip diffs we've already seen (rkey/TID is lexicographically sortable by time)
         if let Some(after) = after_rkey {
             if rkey_str <= after {
-                tracing::debug!("Skipping already-seen diff rkey: {}", rkey_str);
+                tracing::trace!("Skipping already-seen diff rkey: {}", rkey_str);
                 continue;
             }
         }
···
 #[cfg(all(target_arch = "wasm32", target_os = "unknown"))]
 #[component]
 pub fn RealTimeSync(props: RealTimeSyncProps) -> Element {
+    use crate::collab_context::try_use_collab_debug;
     use tokio::sync::mpsc;
+    use weaver_common::WeaverExt;
     use weaver_common::transport::{CollabMessage, CollabSession, SessionEvent};
-    use weaver_common::WeaverExt;
 
     let collab_node = use_collab_node();
     let fetcher = use_context::<crate::fetch::Fetcher>();
     let mut session: Signal<Option<Arc<CollabSession>>> = use_signal(|| None);
     // URI of our published session record (for cleanup)
     let mut session_record_uri: Signal<Option<AtUri<'static>>> = use_signal(|| None);
+    // Debug state for display in editor debug panel (optional, may not be provided)
+    let debug_state = try_use_collab_debug();
     // Channel for sending local updates from Loro callback to async broadcast task
     let mut update_tx: Signal<Option<mpsc::UnboundedSender<Vec<u8>>>> = use_signal(|| None);
     // Channel for sending cursor updates
···
             cursor_tx.set(Some(ctx));
 
             // Subscribe to local updates from Loro - fires when local changes are committed
-            let sub = doc_for_join.loro_doc().subscribe_local_update(Box::new(move |update| {
-                tracing::debug!("RealTimeSync: local update ({} bytes)", update.len());
-                if let Err(e) = tx.send(update.to_vec()) {
-                    tracing::warn!("RealTimeSync: failed to queue update: {}", e);
-                }
-                true // Keep subscription active
-            }));
+            let sub = doc_for_join
+                .loro_doc()
+                .subscribe_local_update(Box::new(move |update| {
+                    tracing::debug!("RealTimeSync: local update ({} bytes)", update.len());
+                    if let Err(e) = tx.send(update.to_vec()) {
+                        tracing::warn!("RealTimeSync: failed to queue update: {}", e);
+                    }
+                    true // Keep subscription active
+                }));
             _subscription.set(Some(sub));
 
             let doc_for_recv = doc_for_join.clone();
···
                 // Derive topic from resource URI
                 let topic = CollabSession::topic_from_uri(uri.as_str());
 
+                // Wait for relay connection before discovering peers or publishing session
+                // Browser clients REQUIRE relay for peer connectivity
+                let relay_url = node.wait_for_relay().await;
+                tracing::info!(
+                    relay_url = %relay_url,
+                    "RealTimeSync: relay connection ready"
+                );
+
+                // Update debug state with node info
+                if let Some(mut ds) = debug_state {
+                    ds.with_mut(|s| {
+                        s.node_id = Some(node.node_id_string());
+                        s.relay_url = Some(relay_url.clone());
+                    });
+                }
+
                 // Discover existing session peers for bootstrap
                 let bootstrap_peers = match fetcher.find_session_peers(&uri).await {
                     Ok(peers) => {
                         tracing::info!("RealTimeSync: found {} existing peers", peers.len());
+                        if let Some(mut ds) = debug_state {
+                            ds.with_mut(|s| s.discovered_peers = peers.len());
+                        }
+                        for p in &peers {
+                            tracing::info!(
+                                did = %p.did,
+                                node_id = %p.node_id,
+                                relay_url = ?p.relay_url,
+                                expires_at = ?p.expires_at,
+                                "RealTimeSync: discovered peer"
+                            );
+                        }
                         peers
                             .into_iter()
                             .filter_map(|p| {
···
                     }
                     Err(e) => {
                         tracing::warn!("RealTimeSync: failed to find peers: {}", e);
+                        if let Some(mut ds) = debug_state {
+                            ds.with_mut(|s| s.last_error = Some(format!("peer discovery: {}", e)));
+                        }
                         vec![]
                     }
                 };
···
                     .create_collab_session(
                         &resource_ref_for_spawn,
                         &node_id_str,
-                        None, // relay_url - could add if needed
+                        Some(&relay_url),
                         Some(SESSION_TTL_MINUTES),
                     )
                     .await
                 {
                     Ok(uri) => {
                         tracing::info!("RealTimeSync: published session record: {}", uri);
+                        if let Some(mut ds) = debug_state {
+                            ds.with_mut(|s| s.session_record_uri = Some(uri.to_string()));
+                        }
                         session_record_uri.set(Some(uri));
                     }
                     Err(e) => {
                         tracing::warn!("RealTimeSync: failed to publish session record: {}", e);
+                        if let Some(mut ds) = debug_state {
+                            ds.with_mut(|s| s.last_error = Some(format!("publish session: {}", e)));
+                        }
                     }
                 }
 
+                // Clone before join() consumes them
+                let node_for_discovery = node.clone();
+                let bootstrap_peers_set = bootstrap_peers.clone();
+
                 match CollabSession::join(node, topic, bootstrap_peers).await {
                     Ok((collab_session, mut event_stream)) => {
                         let collab_session = Arc::new(collab_session);
                         session.set(Some(collab_session.clone()));
+                        if let Some(mut ds) = debug_state {
+                            ds.with_mut(|s| s.is_joined = true);
+                        }
 
                         tracing::info!("RealTimeSync: joined session for {}", uri);
 
+                        // Broadcast Join message to announce ourselves
+                        let our_did = fetcher.current_did().await;
+                        let display_name = if let Some(ref did) = our_did {
+                            use jacquard::types::ident::AtIdentifier;
+                            use weaver_api::sh_weaver::actor::ProfileDataViewInner;
+
+                            let ident = AtIdentifier::Did(did.clone());
+                            fetcher
+                                .fetch_profile(&ident)
+                                .await
+                                .ok()
+                                .and_then(|p| match &p.inner {
+                                    ProfileDataViewInner::ProfileView(pv) => {
+                                        pv.display_name.as_ref().map(|s| s.to_string())
+                                    }
+                                    ProfileDataViewInner::ProfileViewDetailed(pv) => {
+                                        pv.display_name.as_ref().map(|s| s.to_string())
+                                    }
+                                    _ => None,
+                                })
+                                .unwrap_or_else(|| "Collaborator".to_string())
+                        } else {
+                            "Collaborator".to_string()
+                        };
+                        let join_msg = CollabMessage::Join {
+                            did: our_did.map(|d| d.to_string()).unwrap_or_default(),
+                            display_name,
+                        };
+                        if let Err(e) = collab_session.broadcast(&join_msg).await {
+                            tracing::warn!("RealTimeSync: failed to broadcast Join: {}", e);
+                        }
+
+                        // Request sync from existing peers
+                        // Convert our version vector to wire format
+                        let our_vv = doc_for_recv.version_vector();
+                        let have_version: Vec<(u64, u64)> = our_vv
+                            .iter()
+                            .map(|(peer, counter)| (*peer, *counter as u64))
+                            .collect();
+                        let sync_request = CollabMessage::SyncRequest { have_version };
+                        if let Err(e) = collab_session.broadcast(&sync_request).await {
+                            tracing::warn!("RealTimeSync: failed to broadcast SyncRequest: {}", e);
+                        } else {
+                            tracing::debug!("RealTimeSync: sent sync request ({} vv entries)", our_vv.len());
+                        }
+
+                        // Spawn TTL refresh task - keeps our session record alive
+                        let session_uri_for_refresh = session_record_uri.clone();
+                        let fetcher_for_refresh = fetcher.clone();
+                        spawn(async move {
+                            // Refresh every 5 minutes (TTL is 15 min, so plenty of buffer)
+                            let mut interval = n0_future::time::interval(
+                                n0_future::time::Duration::from_secs(5 * 60),
+                            );
+                            loop {
+                                interval.tick().await;
+                                if let Some(uri) = session_uri_for_refresh.peek().clone() {
+                                    tracing::debug!("RealTimeSync: refreshing session TTL");
+                                    if let Err(e) = fetcher_for_refresh
+                                        .refresh_collab_session(&uri, SESSION_TTL_MINUTES)
+                                        .await
+                                    {
+                                        tracing::warn!(
+                                            "RealTimeSync: failed to refresh session: {}",
+                                            e
+                                        );
+                                    }
+                                }
+                            }
+                        });
+
                         // Spawn broadcast task - sends local updates to gossip
                         let session_for_broadcast = collab_session.clone();
                         spawn(async move {
···
                             }
                         });
 
+                        // Spawn periodic peer discovery task
+                        // This handles the race condition where peers publish sessions
+                        // at different times and might miss each other on initial discovery
+                        let session_for_discovery = collab_session.clone();
+                        let fetcher_for_discovery = fetcher.clone();
+                        let uri_for_discovery = uri.clone();
+                        let our_node_id = node_for_discovery.node_id();
+                        let mut known_peers: std::collections::HashSet<weaver_common::transport::EndpointId> =
+                            bootstrap_peers_set.iter().cloned().collect();
+                        spawn(async move {
+                            // Check for new peers every 30 seconds
+                            let mut interval =
+                                n0_future::time::interval(n0_future::time::Duration::from_secs(30));
+                            loop {
+                                interval.tick().await;
+                                tracing::debug!("RealTimeSync: periodic discovery tick");
+                                match fetcher_for_discovery
+                                    .find_session_peers(&uri_for_discovery)
+                                    .await
+                                {
+                                    Ok(peers) => {
+                                        tracing::info!(
+                                            "RealTimeSync: periodic discovery found {} session records",
+                                            peers.len()
+                                        );
+                                        for p in &peers {
+                                            tracing::debug!(
+                                                "  - peer: {} (relay: {:?}, expires: {:?})",
+                                                p.node_id,
+                                                p.relay_url,
+                                                p.expires_at
+                                            );
+                                        }
+                                        // Filter: parse node ID, exclude ourselves, exclude already known
+                                        let new_peers: Vec<_> = peers
+                                            .into_iter()
+                                            .filter_map(|p| {
+                                                weaver_common::transport::parse_node_id(&p.node_id)
+                                                    .ok()
+                                            })
+                                            .filter(|id| *id != our_node_id)
+                                            .filter(|id| !known_peers.contains(id))
+                                            .collect();
+
+                                        if !new_peers.is_empty() {
+                                            tracing::info!(
+                                                "RealTimeSync: periodic discovery found {} NEW peers",
+                                                new_peers.len()
+                                            );
+                                            for p in &new_peers {
+                                                known_peers.insert(*p);
+                                            }
+                                            if let Err(e) =
+                                                session_for_discovery.join_peers(new_peers).await
+                                            {
+                                                tracing::warn!(
+                                                    "RealTimeSync: failed to join discovered peers: {}",
+                                                    e
+                                                );
+                                            }
+                                        }
+                                    }
+                                    Err(e) => {
+                                        tracing::warn!(
+                                            "RealTimeSync: periodic peer discovery failed: {}",
+                                            e
+                                        );
+                                    }
+                                }
+                            }
+                        });
+
                         // Spawn event receiver task - receives updates from peers
                         let mut doc_for_recv = doc_for_recv.clone();
+                        let session_for_sync = collab_session.clone();
                         spawn(async move {
                             use n0_future::StreamExt;
 
-                            while let Some(event) = event_stream.next().await {
+                            while let Some(result) = event_stream.next().await {
+                                let event = match result {
+                                    Ok(e) => e,
+                                    Err(e) => {
+                                        tracing::error!("RealTimeSync: event stream error: {}", e);
+                                        break;
+                                    }
+                                };
                                 match event {
                                     SessionEvent::Message { from, message } => {
                                         match message {
···
                                                 selection,
                                                 ..
                                             } => {
-                                                presence.write().update_cursor(
-                                                    &from,
-                                                    position,
-                                                    selection,
-                                                );
+                                                // Add peer if not known (cursor might arrive before Join)
+                                                let mut p = presence.write();
+                                                if !p.contains(&from) {
+                                                    p.add_collaborator(
+                                                        from,
+                                                        "unknown".into(),
+                                                        "Peer".into(),
+                                                    );
+                                                }
+                                                p.update_cursor(&from, position, selection);
                                             }
                                             CollabMessage::Join { did, display_name } => {
                                                 tracing::info!(
···
                                                     "RealTimeSync: sync request (have {} entries)",
                                                     have_version.len()
                                                 );
-                                                // TODO: Send snapshot or updates based on their version
+                                                // Convert their version vector from wire format
+                                                let their_vv: loro::VersionVector = have_version
+                                                    .into_iter()
+                                                    .map(|(peer, counter)| (peer, counter as i32))
+                                                    .collect();
+
+                                                // Export updates they don't have
+                                                if let Some(data) =
+                                                    doc_for_recv.export_updates_from(&their_vv)
+                                                {
+                                                    tracing::info!(
+                                                        "RealTimeSync: sending {} bytes to sync peer",
+                                                        data.len()
+                                                    );
+                                                    let response = CollabMessage::SyncResponse {
+                                                        data,
+                                                        is_snapshot: false,
+                                                    };
+                                                    if let Err(e) =
+                                                        session_for_sync.broadcast(&response).await
+                                                    {
+                                                        tracing::warn!(
+                                                            "RealTimeSync: failed to send sync response: {}",
+                                                            e
+                                                        );
+                                                    }
+                                                } else {
+                                                    tracing::debug!(
+                                                        "RealTimeSync: no updates to send (peer is up to date)"
+                                                    );
+                                                }
                                             }
-                                            _ => {}
+                                            CollabMessage::SyncResponse { data, is_snapshot } => {
+                                                tracing::info!(
+                                                    "RealTimeSync: received sync response ({} bytes, snapshot: {})",
+                                                    data.len(),
+                                                    is_snapshot
+                                                );
+                                                if let Err(e) = doc_for_recv.import_updates(&data) {
+                                                    tracing::warn!(
+                                                        "RealTimeSync: failed to import sync response: {:?}",
+                                                        e
+                                                    );
+                                                }
+                                            }
                                         }
                                     }
                                     SessionEvent::PeerJoined(peer) => {

crates/weaver-app/src/main.rs

···
         );
 
         // Filter out noisy crates
-        let filter = EnvFilter::new("debug,loro_internal=warn");
+        let filter =
+            EnvFilter::new("debug,loro_internal=warn,jacquard_identity=info,jacquard_common=info");
 
         let reg = Registry::default()
             .with(filter)

crates/weaver-common/Cargo.toml

···
 send_wrapper = "0.6"
 wasmworker = "0.1"
 wasmworker-proc-macro = "0.1"
-iroh-quinn = { version = "0.14", default-features = false }
 ring = { version = "0.17", default-features = false, features = ["wasm32_unknown_unknown_js"]}
 getrandom = { version = "0.3", default-features = false, features = ["wasm_js"] }
 

crates/weaver-common/src/agent.rs

···
                 };
 
                 if !accept_output.records.is_empty() {
+                    // Both parties in a valid invite+accept pair are authorized
+                    let inviter_did = record_id.did.clone().into_static();
+                    collaborators.push(inviter_did);
                     collaborators.push(invitee_did);
                 }
             }
+
+            // Deduplicate (someone might appear in multiple pairs)
+            collaborators.sort();
+            collaborators.dedup();
 
             Ok(collaborators)
         }
···
             use jacquard::types::string::Datetime;
             use weaver_api::sh_weaver::collab::session::Session;
 
+            // Clean up any expired sessions first
+            let _ = self.cleanup_expired_sessions().await;
+
             let now_chrono = chrono::Utc::now().fixed_offset();
             let now = Datetime::new(now_chrono);
             let expires_at = ttl_minutes.map(|mins| {
···
         }
     }
 
+    /// Update the relay URL in an existing session record.
+    ///
+    /// Called when the relay connection changes during a session.
+    fn update_collab_session_relay<'a>(
+        &'a self,
+        session_uri: &'a AtUri<'a>,
+        relay_url: Option<&'a str>,
+    ) -> impl Future<Output = Result<(), WeaverError>> + 'a {
+        async move {
+            use weaver_api::sh_weaver::collab::session::Session;
+
+            let relay_uri = relay_url
+                .map(|url| jacquard::types::string::Uri::new(url))
+                .transpose()
+                .map_err(|_| AgentError::from(ClientError::invalid_request("Invalid relay URL")))?;
+
+            self.update_record::<Session>(session_uri, |session| {
+                session.relay_url = relay_uri.clone();
+            })
+            .await?;
+            Ok(())
+        }
+    }
+
+    /// Delete all expired session records for the current user.
+    ///
+    /// Called before creating a new session to clean up stale records.
+    fn cleanup_expired_sessions<'a>(
+        &'a self,
+    ) -> impl Future<Output = Result<u32, WeaverError>> + 'a
+    where
+        Self: Sized,
+    {
+        async move {
+            use jacquard::types::nsid::Nsid;
+            use weaver_api::com_atproto::repo::list_records::ListRecords;
+            use weaver_api::sh_weaver::collab::session::Session;
+
+            let (did, _) = self.session_info().await.ok_or_else(|| {
+                AgentError::from(ClientError::invalid_request("No active session"))
+            })?;
+            let now = chrono::Utc::now();
+            let mut deleted = 0u32;
+
+            // List all our session records
+            let collection =
+                Nsid::new("sh.weaver.collab.session").map_err(WeaverError::AtprotoString)?;
+            let request = ListRecords::new()
+                .repo(did.clone())
+                .collection(collection)
+                .limit(100)
+                .build();
+
+            let response = self.send(request).await.map_err(AgentError::from)?;
+            let output = response.into_output().map_err(|e| {
+                AgentError::from(ClientError::invalid_request(format!(
+                    "Failed to list sessions: {}",
+                    e
+                )))
+            })?;
+
+            for record in output.records {
+                if let Ok(session) = jacquard::from_data::<Session>(&record.value) {
+                    // Check if expired
+                    if let Some(ref expires_at) = session.expires_at {
+                        let expires_str = expires_at.as_str();
+                        if let Ok(expires) = chrono::DateTime::parse_from_rfc3339(expires_str) {
+                            if expires.with_timezone(&chrono::Utc) < now {
+                                // Delete expired session
+                                if let Some(rkey) = record.uri.rkey() {
+                                    if let Err(e) =
+                                        self.delete_record::<Session>(rkey.clone()).await
+                                    {
+                                        tracing::warn!("Failed to delete expired session: {}", e);
+                                    } else {
+                                        deleted += 1;
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+
+            if deleted > 0 {
+                tracing::info!("Cleaned up {} expired session records", deleted);
+            }
+
+            Ok(deleted)
+        }
+    }
+
     /// Find active collaboration sessions for a resource.
     ///
     /// Queries Constellation for session records referencing the given resource,
···
             use weaver_api::sh_weaver::collab::session::Session;
 
             const SESSION_NSID: &str = "sh.weaver.collab.session";
+
+            // Get authorized collaborators (owner is checked separately via URI authority)
+            let collaborators: std::collections::HashSet<Did<'static>> = self
+                .find_collaborators_for_resource(resource_uri)
+                .await
+                .unwrap_or_default()
+                .into_iter()
+                .collect();
 
             let constellation_url = Url::parse(CONSTELLATION_URL).map_err(|e| {
                 AgentError::from(ClientError::invalid_request(format!(
···
                     if *expires_at < now {
                         continue; // Session expired
                     }
+                }
+
+                // Check if peer is authorized (has valid invite+accept pair)
+                let peer_did = record_id.did.clone().into_static();
+                if !collaborators.contains(&peer_did) {
+                    tracing::debug!(
+                        peer = %peer_did,
+                        "Filtering out unauthorized session peer"
+                    );
+                    continue;
                 }
 
                 peers.push(SessionPeer {

crates/weaver-common/src/transport/messages.rs

···
 //! Wire protocol for collaborative editing messages.
 
+use iroh::{PublicKey, SecretKey, Signature};
 use serde::{Deserialize, Serialize};
 
 /// Messages exchanged between collaborators over gossip.
···
 }
 
 impl CollabMessage {
-    /// Serialize message to CBOR bytes for wire transmission
+    /// Serialize message to postcard bytes for wire transmission.
     pub fn to_bytes(&self) -> Result<Vec<u8>, postcard::Error> {
         postcard::to_stdvec(self)
     }
 
-    /// Deserialize message from CBOR bytes
+    /// Deserialize message from postcard bytes.
     pub fn from_bytes(bytes: &[u8]) -> Result<Self, postcard::Error> {
         postcard::from_bytes(bytes)
+    }
+}
+
+/// A signed message wrapper for authenticated transport.
+///
+/// Includes the sender's public key so receivers can verify without context.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct SignedMessage {
+    /// Sender's public key (also their EndpointId).
+    pub from: PublicKey,
+    /// The serialized TimestampedMessage (postcard bytes).
+    pub data: Vec<u8>,
+    /// Ed25519 signature over data.
+    pub signature: Signature,
+}
+
+/// Versioned wire format with timestamp.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+enum WireMessage {
+    V0 { timestamp: u64, message: CollabMessage },
+}
+
+/// A verified message with sender and timestamp info.
+#[derive(Debug, Clone)]
+pub struct ReceivedMessage {
+    /// Sender's public key.
+    pub from: PublicKey,
+    /// When the message was sent (micros since epoch).
+    pub timestamp: u64,
+    /// The decoded message.
+    pub message: CollabMessage,
+}
+
+/// Error type for signed message operations.
+#[derive(Debug, thiserror::Error)]
+pub enum SignedMessageError {
+    #[error("serialization failed: {0}")]
+    Serialization(#[from] postcard::Error),
+    #[error("signature verification failed")]
+    InvalidSignature,
+}
+
+impl SignedMessage {
+    /// Sign a message and encode to bytes for wire transmission.
+    pub fn sign_and_encode(secret_key: &SecretKey, message: &CollabMessage) -> Result<Vec<u8>, SignedMessageError> {
+        use web_time::SystemTime;
+
+        let timestamp = SystemTime::now()
+            .duration_since(SystemTime::UNIX_EPOCH)
+            .unwrap()
+            .as_micros() as u64;
+        let wire = WireMessage::V0 { timestamp, message: message.clone() };
+        let data = postcard::to_stdvec(&wire)?;
+        let signature = secret_key.sign(&data);
+        let from = secret_key.public();
+        let signed = Self { from, data, signature };
+        Ok(postcard::to_stdvec(&signed)?)
+    }
+
+    /// Decode from bytes and verify signature.
+    pub fn decode_and_verify(bytes: &[u8]) -> Result<ReceivedMessage, SignedMessageError> {
+        let signed: Self = postcard::from_bytes(bytes)?;
+        signed.from
+            .verify(&signed.data, &signed.signature)
+            .map_err(|_| SignedMessageError::InvalidSignature)?;
+        let wire: WireMessage = postcard::from_bytes(&signed.data)?;
+        let WireMessage::V0 { timestamp, message } = wire;
+        Ok(ReceivedMessage {
+            from: signed.from,
+            timestamp,
+            message,
+        })
     }
 }
 

crates/weaver-common/src/transport/mod.rs

···
 
 pub use discovery::{node_id_to_string, parse_node_id, DiscoveredPeer, DiscoveryError};
 pub use iroh::EndpointId;
-pub use messages::CollabMessage;
+pub use messages::{CollabMessage, ReceivedMessage, SignedMessage, SignedMessageError};
 pub use node::{CollabNode, TransportError};
 pub use presence::{Collaborator, PresenceTracker, RemoteCursor};
 pub use session::{CollabSession, SessionError, SessionEvent, TopicId};
-
-/// ALPN protocol identifier for weaver collaboration
-pub const WEAVER_GOSSIP_ALPN: &[u8] = b"weaver/collab/0";

crates/weaver-common/src/transport/node.rs

···
 use iroh::Endpoint;
 use iroh::EndpointId;
 use iroh::SecretKey;
-use iroh_gossip::net::Gossip;
+use iroh_gossip::net::{GOSSIP_ALPN, Gossip};
 use miette::Diagnostic;
 use std::sync::Arc;
-
-use super::WEAVER_GOSSIP_ALPN;
 
 /// Error type for transport operations
 #[derive(Debug, thiserror::Error, Diagnostic)]
···
         // In native, this can do direct P2P with relay fallback
         let endpoint = Endpoint::builder()
             .secret_key(secret_key.clone())
-            .alpns(vec![WEAVER_GOSSIP_ALPN.to_vec()])
+            .alpns(vec![GOSSIP_ALPN.to_vec()])
             .bind()
             .await
             .map_err(|e| TransportError::Bind(Box::new(e)))?;
···
 
         // Build router to dispatch incoming connections by ALPN
         let router = iroh::protocol::Router::builder(endpoint.clone())
-            .accept(WEAVER_GOSSIP_ALPN, gossip.clone())
+            .accept(GOSSIP_ALPN, gossip.clone())
             .spawn();
 
         tracing::info!(node_id = %endpoint.id(), "CollabNode started");
···
     /// Get a clone of the secret key (for session persistence if needed).
     pub fn secret_key(&self) -> SecretKey {
         self.secret_key.clone()
+    }
+
+    /// Get the relay URL this node is connected to (if any).
+    ///
+    /// This should be published in session records so other peers can connect
+    /// via relay (essential for browser-to-browser connections).
+    pub fn relay_url(&self) -> Option<String> {
+        self.endpoint
+            .addr()
+            .relay_urls()
+            .next()
+            .map(|url| url.to_string())
+    }
+
+    /// Get the full node address including relay info.
+    ///
+    /// Use this when you need to connect to this node from another peer.
+    pub fn node_addr(&self) -> iroh::EndpointAddr {
+        self.endpoint.addr()
+    }
+
+    /// Wait for the endpoint to be online (relay connected).
+    ///
+    /// This should be called before publishing session records to ensure
+    /// the relay URL is available for peer discovery. For browser clients,
+    /// relay is required - we wait indefinitely since there's no fallback.
+    pub async fn wait_online(&self) {
+        self.endpoint.online().await;
+    }
+
+    /// Wait for relay connection and return the relay URL.
+    ///
+    /// Waits indefinitely for relay - browser clients require relay URLs
+    /// for peer discovery. Returns the relay URL once connected.
+    pub async fn wait_for_relay(&self) -> String {
+        self.endpoint.online().await;
+        // After online(), relay_url should always be Some for browser clients
+        self.relay_url()
+            .expect("relay URL should be available after online()")
+    }
+
+    /// Watch for address changes (including relay URL changes).
+    ///
+    /// Returns a stream that yields the address on each change.
+    /// Use this to detect relay URL changes and update session records.
+    pub fn watch_addr(&self) -> n0_future::boxed::BoxStream<iroh::EndpointAddr> {
+        use iroh::Watcher;
+        Box::pin(self.endpoint.watch_addr().stream())
     }
 }

crates/weaver-common/src/transport/session.rs

···
 use n0_future::boxed::BoxStream;
 use n0_future::stream;
 
-use super::{CollabMessage, CollabNode};
+use super::{CollabMessage, CollabNode, SignedMessage};
 
 /// Topic ID for a gossip session - derived from resource URI.
 pub type TopicId = iroh_gossip::TopicId;
···
 pub struct CollabSession {
     topic: TopicId,
     sender: GossipSender,
-    #[allow(dead_code)]
     node: Arc<CollabNode>,
 }
 
···
         node: Arc<CollabNode>,
         topic: TopicId,
         bootstrap_peers: Vec<EndpointId>,
-    ) -> Result<(Self, BoxStream<SessionEvent>), SessionError> {
+    ) -> Result<(Self, BoxStream<Result<SessionEvent, SessionError>>), SessionError> {
+        tracing::info!(
+            topic = ?topic,
+            bootstrap_count = bootstrap_peers.len(),
+            "CollabSession: joining topic"
+        );
+
+        for peer in &bootstrap_peers {
+            tracing::debug!(peer = %peer, "CollabSession: bootstrap peer");
+        }
+
         // Subscribe to the gossip topic
         let (sender, receiver) = node
             .gossip()
-            .subscribe(topic, bootstrap_peers)
+            .subscribe_and_join(topic, bootstrap_peers)
             .await
             .map_err(|e| SessionError::Subscribe(Box::new(e)))?
             .split();
+
+        tracing::info!("CollabSession: subscribed to gossip topic");
 
         let session = Self {
             topic,
···
     }
 
     /// Convert gossip receiver into a stream of session events.
-    fn event_stream(receiver: GossipReceiver) -> BoxStream<SessionEvent> {
-        let stream = stream::unfold(receiver, |mut receiver| async move {
+    fn event_stream(receiver: GossipReceiver) -> BoxStream<Result<SessionEvent, SessionError>> {
+        let stream = stream::try_unfold(receiver, |mut receiver| async move {
             loop {
-                match receiver.next().await {
-                    Some(Ok(event)) => {
-                        let session_event = match event {
-                            Event::NeighborUp(peer) => SessionEvent::PeerJoined(peer),
-                            Event::NeighborDown(peer) => SessionEvent::PeerLeft(peer),
-                            Event::Received(msg) => match CollabMessage::from_bytes(&msg.content) {
-                                Ok(message) => SessionEvent::Message {
-                                    from: msg.delivered_from,
-                                    message,
-                                },
-                                Err(e) => {
-                                    tracing::warn!(?e, "failed to decode collab message");
+                let Some(event) = receiver.try_next().await.map_err(|e| {
+                    tracing::error!(?e, "CollabSession: gossip receiver error");
+                    SessionError::Decode(Box::new(e))
+                })?
+                else {
+                    tracing::debug!("CollabSession: gossip stream ended");
+                    return Ok(None);
+                };
+
+                tracing::debug!(?event, "CollabSession: raw gossip event");
+                let session_event = match event {
+                    Event::NeighborUp(peer) => {
+                        tracing::info!(peer = %peer, "CollabSession: neighbor up");
+                        SessionEvent::PeerJoined(peer)
+                    }
+                    Event::NeighborDown(peer) => {
+                        tracing::info!(peer = %peer, "CollabSession: neighbor down");
+                        SessionEvent::PeerLeft(peer)
+                    }
+                    Event::Received(msg) => {
+                        tracing::debug!(
+                            from = %msg.delivered_from,
+                            bytes = msg.content.len(),
+                            "CollabSession: received message"
+                        );
+                        match SignedMessage::decode_and_verify(&msg.content) {
+                            Ok(received) => {
+                                // Verify claimed sender matches transport sender
+                                if received.from != msg.delivered_from {
+                                    tracing::warn!(
+                                        claimed = %received.from,
+                                        transport = %msg.delivered_from,
+                                        "sender mismatch - possible spoofing attempt"
+                                    );
                                     continue;
                                 }
-                            },
-                            Event::Lagged => {
-                                tracing::warn!("gossip receiver lagged, some messages may be lost");
+                                SessionEvent::Message {
+                                    from: received.from,
+                                    message: received.message,
+                                }
+                            }
+                            Err(e) => {
+                                tracing::warn!(?e, "failed to verify/decode signed message");
                                 continue;
                             }
-                        };
-                        return Some((session_event, receiver));
+                        }
                     }
-                    Some(Err(e)) => {
-                        tracing::warn!(?e, "gossip receiver error");
+                    Event::Lagged => {
+                        tracing::warn!("gossip receiver lagged, some messages may be lost");
                         continue;
                     }
-                    None => return None,
-                }
+                };
+                break Ok(Some((session_event, receiver)));
             }
         });
 
         Box::pin(stream)
     }
 
-    /// Broadcast a message to all peers in the session.
+    /// Broadcast a signed message to all peers in the session.
     pub async fn broadcast(&self, message: &CollabMessage) -> Result<(), SessionError> {
-        let bytes = message
-            .to_bytes()
+        let bytes = SignedMessage::sign_and_encode(&self.node.secret_key(), message)
             .map_err(|e| SessionError::Broadcast(Box::new(e)))?;
+
+        tracing::debug!(
+            bytes = bytes.len(),
+            topic = ?self.topic,
+            "CollabSession: broadcasting signed message"
+        );
 
         self.sender
             .broadcast(bytes.into())
···
     /// Get the topic ID for this session.
     pub fn topic(&self) -> TopicId {
         self.topic
+    }
+
+    /// Add new peers to the gossip session.
+    ///
+    /// Use this to add peers discovered after initial subscription.
+    /// The gossip layer will attempt to connect to these peers.
+    pub async fn join_peers(&self, peers: Vec<EndpointId>) -> Result<(), SessionError> {
+        if peers.is_empty() {
+            return Ok(());
+        }
+        tracing::info!(
+            count = peers.len(),
+            "CollabSession: joining additional peers"
+        );
+        for peer in &peers {
+            tracing::debug!(peer = %peer, "CollabSession: adding peer");
+        }
+        self.sender
+            .join_peers(peers)
+            .await
+            .map_err(|e| SessionError::Subscribe(Box::new(e)))?;
+        Ok(())
     }
 }