fix(mobile): avoid opening empty threads on message tap (#7756)

Tapping a mobile channel message with no replies currently opens an
empty thread. Only enable message-row thread navigation when the
existing thread summary reports replies, including replies known to the
relay but not loaded locally. Long-press → Reply still starts a new
thread.

Added widget tests for zero replies, loaded replies, relay-only replies,
and starting a thread through Reply. The zero-reply test failed before
the fix and passes afterward.

Validation: all 227 channel-page tests, the full mobile pre-push checks,
and all PR CI checks passed. The full repository gate failed in
unchanged Rust ACP timing tests (idle reset, keepalive, and steering
deadlines); this PR changes only mobile Dart files.

No duplicate open PR or issue found.

---------

Signed-off-by: Tom Brow <tomb@block.xyz>

.github/workflows/_ci-relay.yml

@@ -587,7 +587,7 @@ jobs:
           ./scripts/start-relay-for-tests.sh --no-build
       - name: Relay E2E tests
         run: |
-          cargo test -p buzz-test-client --test e2e_persona --test e2e_team_catalog --test e2e_nostr_interop --test e2e_project -- --ignored --nocapture
+          cargo test -p buzz-test-client --test e2e_persona --test e2e_team_catalog --test e2e_nostr_interop --test e2e_thread_roots --test e2e_project -- --ignored --nocapture
           cargo test -p buzz-test-client --test e2e_relay invite -- --ignored --nocapture
           cargo test -p buzz-test-client --test e2e_relay nip43_membership_snapshots_are_rejected -- --ignored --nocapture
           cargo test -p buzz-test-client --test e2e_relay nip29_departure_wire -- --ignored --nocapture

crates/buzz-db/src/store/thread.rs

@@ -405,7 +405,7 @@ pub(crate) async fn get_thread_replies_on(
     // Decode cursor bytes -> keyset (timestamp, optional event_id) for the
     // WHERE condition. Layout: 8-byte BE i64 seconds, then the raw event_id.
     // An 8-byte-only cursor is legacy timestamp-only paging (no tiebreak).
-    let cursor_key: Option<(DateTime<Utc>, Option<Vec<u8>>)> = match cursor {
+    let mut cursor_key: Option<(DateTime<Utc>, Option<Vec<u8>>)> = match cursor {
         Some(bytes) if bytes.len() >= 8 => {
             let secs = i64::from_be_bytes(bytes[..8].try_into().expect("length checked"));
             DateTime::from_timestamp(secs, 0).map(|ts| {
@@ -420,11 +420,20 @@ pub(crate) async fn get_thread_replies_on(
         _ => None,
     };
 
-    // Build the query dynamically based on optional filters.
-    // Track the next positional parameter index.
-    let mut param_idx = 3u32; // $1 is community_id, $2 is root_event_id
-    let mut sql = String::from(
-        r#"
+    let mut replies = Vec::new();
+    // Damaged rows must not cause unbounded work on a held writer connection.
+    // Fail explicitly rather than exposing a partial page as authoritative EOF.
+    const MAX_RAW_PAGES: usize = 64;
+    for _ in 0..MAX_RAW_PAGES {
+        let remaining = limit.saturating_sub(replies.len() as u32);
+        if remaining == 0 {
+            return Ok(replies);
+        }
+        // Build the query dynamically based on optional filters.
+        // Track the next positional parameter index.
+        let mut param_idx = 3u32; // $1 is community_id, $2 is root_event_id
+        let mut sql = String::from(
+            r#"
         SELECT
             tm.event_id,
             e.id,
@@ -450,91 +459,103 @@ pub(crate) async fn get_thread_replies_on(
           AND tm.root_event_id = $2
           AND e.deleted_at IS NULL
         "#,
-    );
+        );
 
-    if depth_limit.is_some() {
-        sql.push_str(&format!(" AND tm.depth <= ${param_idx}"));
-        param_idx += 1;
-    }
-    match &cursor_key {
-        Some((_, Some(_))) => {
-            // Composite keyset: strict row comparison with an event_id tiebreak
-            // so same-second replies paginate without gaps or duplicates.
-            let ts_idx = param_idx;
-            let id_idx = param_idx + 1;
-            sql.push_str(&format!(
-                " AND (tm.event_created_at, tm.event_id) > (${ts_idx}, ${id_idx})"
-            ));
-            param_idx += 2;
-        }
-        Some((_, None)) => {
-            // Legacy timestamp-only cursor (no tiebreak).
-            sql.push_str(&format!(" AND tm.event_created_at > ${param_idx}"));
+        if depth_limit.is_some() {
+            sql.push_str(&format!(" AND tm.depth <= ${param_idx}"));
             param_idx += 1;
         }
-        None => {}
-    }
-
-    sql.push_str(&format!(
-        " ORDER BY tm.event_created_at ASC, tm.event_id ASC LIMIT ${param_idx}"
-    ));
-
-    let mut q = sqlx::query(sqlx::AssertSqlSafe(sql))
-        .bind(community_id.as_uuid())
-        .bind(root_event_id);
-
-    if let Some(dl) = depth_limit {
-        q = q.bind(dl as i32);
-    }
-    match &cursor_key {
-        Some((ts, Some(id))) => {
-            q = q.bind(*ts).bind(id.clone());
+        match &cursor_key {
+            Some((_, Some(_))) => {
+                // Composite keyset: strict row comparison with an event_id tiebreak
+                // so same-second replies paginate without gaps or duplicates.
+                let ts_idx = param_idx;
+                let id_idx = param_idx + 1;
+                sql.push_str(&format!(
+                    " AND (tm.event_created_at, tm.event_id) > (${ts_idx}, ${id_idx})"
+                ));
+                param_idx += 2;
+            }
+            Some((_, None)) => {
+                // Legacy timestamp-only cursor (no tiebreak).
+                sql.push_str(&format!(" AND tm.event_created_at > ${param_idx}"));
+                param_idx += 1;
+            }
+            None => {}
         }
-        Some((ts, None)) => {
-            q = q.bind(*ts);
+
+        sql.push_str(&format!(
+            " ORDER BY tm.event_created_at ASC, tm.event_id ASC LIMIT ${param_idx}"
+        ));
+
+        let mut q = sqlx::query(sqlx::AssertSqlSafe(sql))
+            .bind(community_id.as_uuid())
+            .bind(root_event_id);
+
+        if let Some(dl) = depth_limit {
+            q = q.bind(dl as i32);
+        }
+        match &cursor_key {
+            Some((ts, Some(id))) => {
+                q = q.bind(*ts).bind(id.clone());
+            }
+            Some((ts, None)) => {
+                q = q.bind(*ts);
+            }
+            None => {}
+        }
+        q = q.bind(remaining as i32);
+
+        let rows = q.fetch_all(&mut *conn).await?;
+
+        let raw_count = rows.len();
+        if let Some(last) = rows.last() {
+            cursor_key = Some((
+                last.try_get("event_created_at")?,
+                Some(last.try_get("event_id")?),
+            ));
+        }
+        for row in rows {
+            let event_id: Vec<u8> = row.try_get("event_id")?;
+            let parent_event_id: Option<Vec<u8>> = row.try_get("parent_event_id")?;
+            let root_event_id_col: Option<Vec<u8>> = row.try_get("root_event_id")?;
+            let channel_id: Uuid = row.try_get("channel_id")?;
+            let pubkey: Vec<u8> = row.try_get("pubkey")?;
+            let tags: serde_json::Value = row.try_get("tags")?;
+            let depth: i32 = row.try_get("depth")?;
+            let created_at: DateTime<Utc> = row.try_get("event_created_at")?;
+            let broadcast_val: bool = row.try_get("broadcast")?;
+
+            // Skip rows that fail event reconstruction (e.g. corrupt signature)
+            // while continuing past the raw page boundary to fill the requested
+            // page. A short response must mean SQL EOF, not a skipped stored row.
+            let stored_event = match row_to_stored_event(row)? {
+                Some(se) => se,
+                None => continue,
+            };
+
+            replies.push(ThreadReply {
+                event_id,
+                parent_event_id,
+                root_event_id: root_event_id_col,
+                channel_id,
+                pubkey,
+                tags,
+                content: stored_event.event.content.clone(),
+                stored_event,
+                depth,
+                created_at,
+                broadcast: broadcast_val,
+            });
+        }
+
+        if raw_count < remaining as usize || replies.len() == limit as usize {
+            return Ok(replies);
         }
-        None => {}
     }
-    q = q.bind(limit as i32);
-
-    let rows = q.fetch_all(&mut *conn).await?;
-
-    let mut replies = Vec::with_capacity(rows.len());
-    for row in rows {
-        let event_id: Vec<u8> = row.try_get("event_id")?;
-        let parent_event_id: Option<Vec<u8>> = row.try_get("parent_event_id")?;
-        let root_event_id_col: Option<Vec<u8>> = row.try_get("root_event_id")?;
-        let channel_id: Uuid = row.try_get("channel_id")?;
-        let pubkey: Vec<u8> = row.try_get("pubkey")?;
-        let tags: serde_json::Value = row.try_get("tags")?;
-        let depth: i32 = row.try_get("depth")?;
-        let created_at: DateTime<Utc> = row.try_get("event_created_at")?;
-        let broadcast_val: bool = row.try_get("broadcast")?;
-
-        // Skip rows that fail event reconstruction (e.g. corrupt signature)
-        // rather than failing the whole thread query, matching the
-        // skip-and-continue semantics of the prior get_events_by_ids path.
-        let stored_event = match row_to_stored_event(row)? {
-            Some(se) => se,
-            None => continue,
-        };
-
-        replies.push(ThreadReply {
-            event_id,
-            parent_event_id,
-            root_event_id: root_event_id_col,
-            channel_id,
-            pubkey,
-            tags,
-            content: stored_event.event.content.clone(),
-            stored_event,
-            depth,
-            created_at,
-            broadcast: broadcast_val,
-        });
-    }
-
-    Ok(replies)
+    Err(crate::error::DbError::InvalidData(
+        "thread page exceeded the corrupt-row refill limit".into(),
+    ))
 }
 
 /// Fetch aggregated thread stats for a single event, plus up to 10 participant pubkeys.
@@ -1043,6 +1064,62 @@ impl Db {
         crate::thread::get_thread_summary(&self.pool, community_id, event_id).await
     }
 
+    /// Resolve channel-scoped reply IDs, including tombstones, to current root
+    /// summaries on the writer. Returns metadata only, never target payloads.
+    #[datastore_span(name = "resolve_thread_root_summaries", system = "postgresql")]
+    pub async fn resolve_thread_root_summaries(
+        &self,
+        community_id: CommunityId,
+        channel_id: Uuid,
+        target_ids: &[Vec<u8>],
+    ) -> Result<Vec<(Vec<u8>, ThreadSummary)>> {
+        let mut connection = crate::observability::acquire_writer(
+            &self.pool,
+            crate::observability::WriterOperation::SubscriptionHistory,
+        )
+        .await?;
+        let rows = sqlx::query(
+            r#"
+            SELECT root.event_id, root.reply_count, root.descendant_count, root.last_reply_at,
+                ARRAY(
+                    SELECT e.pubkey
+                    FROM thread_metadata tm
+                    JOIN events e ON e.community_id = tm.community_id
+                        AND e.created_at = tm.event_created_at AND e.id = tm.event_id
+                    WHERE tm.community_id = root.community_id AND tm.root_event_id = root.event_id
+                        AND tm.channel_id = root.channel_id AND e.deleted_at IS NULL
+                    GROUP BY e.pubkey ORDER BY MAX(e.created_at) DESC LIMIT 10
+                ) AS participants
+            FROM thread_metadata root
+            WHERE root.community_id = $1 AND root.channel_id = $2
+                AND root.event_id IN (
+                    SELECT COALESCE(target.root_event_id, target.parent_event_id)
+                    FROM thread_metadata target
+                    WHERE target.community_id = $1 AND target.channel_id = $2
+                        AND target.event_id = ANY($3) AND target.parent_event_id IS NOT NULL
+                )
+        "#,
+        )
+        .bind(community_id.as_uuid())
+        .bind(channel_id)
+        .bind(target_ids)
+        .fetch_all(&mut *connection)
+        .await?;
+        rows.into_iter()
+            .map(|row| {
+                Ok((
+                    row.try_get("event_id")?,
+                    ThreadSummary {
+                        reply_count: row.try_get("reply_count")?,
+                        descendant_count: row.try_get("descendant_count")?,
+                        last_reply_at: row.try_get("last_reply_at")?,
+                        participants: row.try_get("participants")?,
+                    },
+                ))
+            })
+            .collect()
+    }
+
     /// One channel window: top-level rows + summaries + server `has_more`.
     ///
     /// Convenience wrapper over [`Db::get_channel_window_with_session`] for
@@ -1290,6 +1367,83 @@ mod postgres_tests {
             .map(|channel| (channel, buzz_core::CommunityId::from_uuid(community_id)))
     }
 
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn resolve_deleted_roots_preserves_community_scope_for_colliding_ids() {
+        let pool = setup_pool().await;
+        let db = Db::from_pool(pool.clone());
+        let author = Keys::generate();
+        let channel = Uuid::new_v4();
+        let communities = [
+            CommunityId::from_uuid(make_test_community(&pool).await),
+            CommunityId::from_uuid(make_test_community(&pool).await),
+        ];
+        let root = make_stream_event(&author, "shared root identity");
+        let reply = make_stream_event(&author, "shared reply identity");
+        for community in communities {
+            crate::channel::create_channel_with_id(
+                &pool,
+                community,
+                channel,
+                &format!("root-scope-{channel}"),
+                ChannelType::Stream,
+                ChannelVisibility::Open,
+                None,
+                author.public_key().to_bytes().as_slice(),
+                None,
+            )
+            .await
+            .expect("channel");
+            for (event, parent) in [(&root, None), (&reply, Some(&root))] {
+                insert_event_with_thread_metadata(
+                    &pool,
+                    community,
+                    event,
+                    Some(channel),
+                    Some(ThreadMetadataParams {
+                        event_id: event.id.as_bytes(),
+                        event_created_at: event_created_at(event),
+                        channel_id: channel,
+                        parent_event_id: parent.map(|parent| parent.id.as_bytes().as_slice()),
+                        parent_event_created_at: parent.map(event_created_at),
+                        root_event_id: parent.map(|parent| parent.id.as_bytes().as_slice()),
+                        root_event_created_at: parent.map(event_created_at),
+                        depth: i32::from(parent.is_some()),
+                        broadcast: false,
+                    }),
+                )
+                .await
+                .expect("thread event");
+            }
+        }
+        db.soft_delete_event_and_update_thread(
+            communities[0],
+            reply.id.as_bytes(),
+            Some(root.id.as_bytes()),
+            Some(root.id.as_bytes()),
+        )
+        .await
+        .expect("delete only A");
+        for (community, expected) in [(communities[0], 0), (communities[1], 1)] {
+            let summaries = db
+                .resolve_thread_root_summaries(community, channel, &[reply.id.as_bytes().to_vec()])
+                .await
+                .expect("metadata query");
+            assert_eq!(summaries.len(), 1);
+            assert_eq!(summaries[0].0, root.id.as_bytes());
+            assert_eq!(summaries[0].1.descendant_count, expected);
+        }
+        assert!(db
+            .resolve_thread_root_summaries(
+                communities[0],
+                Uuid::new_v4(),
+                &[reply.id.as_bytes().to_vec()]
+            )
+            .await
+            .expect("wrong channel")
+            .is_empty());
+    }
+
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn get_thread_metadata_by_event_is_scoped_when_event_id_collides_across_communities() {
@@ -1755,7 +1909,10 @@ mod postgres_tests {
             .expect("insert root event");
 
         // Two replies under the same root: one stays valid, one we corrupt.
-        let good = make_stream_event(&author, "good");
+        let good = EventBuilder::new(Kind::Custom(9), "good")
+            .custom_created_at(nostr::Timestamp::from(root.created_at.as_secs() + 2))
+            .sign_with_keys(&author)
+            .expect("good reply");
         let good_id = good.id.to_hex();
         let good_created_at = event_created_at(&good);
         insert_event_with_thread_metadata(
@@ -1778,7 +1935,10 @@ mod postgres_tests {
         .await
         .expect("insert good reply");
 
-        let bad = make_stream_event(&author, "bad");
+        let bad = EventBuilder::new(Kind::Custom(9), "bad")
+            .custom_created_at(nostr::Timestamp::from(root.created_at.as_secs() + 1))
+            .sign_with_keys(&author)
+            .expect("bad reply");
         let bad_created_at = event_created_at(&bad);
         insert_event_with_thread_metadata(
             &pool,
@@ -1821,6 +1981,156 @@ mod postgres_tests {
         assert_eq!(replies.len(), 1);
         assert_eq!(replies[0].stored_event.event.id.to_hex(), good_id);
         assert_eq!(replies[0].stored_event.event.content, "good");
+
+        let tail = EventBuilder::new(Kind::Custom(9), "tail")
+            .custom_created_at(nostr::Timestamp::from(root.created_at.as_secs() + 3))
+            .sign_with_keys(&author)
+            .expect("tail reply");
+        insert_event_with_thread_metadata(
+            &pool,
+            community,
+            &tail,
+            Some(channel.id),
+            Some(ThreadMetadataParams {
+                event_id: tail.id.as_bytes(),
+                event_created_at: event_created_at(&tail),
+                channel_id: channel.id,
+                parent_event_id: Some(root.id.as_bytes()),
+                parent_event_created_at: Some(root_created_at),
+                root_event_id: Some(root.id.as_bytes()),
+                root_event_created_at: Some(root_created_at),
+                depth: 1,
+                broadcast: false,
+            }),
+        )
+        .await
+        .expect("insert tail");
+        let mut origin = (-1_i64).to_be_bytes().to_vec();
+        origin.extend_from_slice(&[0; 32]);
+        let page = get_thread_replies(
+            &pool,
+            community,
+            root.id.as_bytes(),
+            Some(10),
+            2,
+            Some(&origin),
+        )
+        .await
+        .expect("fill page past corrupt first row");
+        assert_eq!(
+            page.iter()
+                .map(|reply| reply.stored_event.event.id)
+                .collect::<Vec<_>>(),
+            [good.id, tail.id]
+        );
+        let first = get_thread_replies(
+            &pool,
+            community,
+            root.id.as_bytes(),
+            Some(10),
+            1,
+            Some(&origin),
+        )
+        .await
+        .expect("page entirely corrupt must not signal EOF");
+        assert_eq!(first[0].stored_event.event.id, good.id);
+        let mut after = (tail.created_at.as_secs() as i64).to_be_bytes().to_vec();
+        after.extend_from_slice(tail.id.as_bytes());
+        assert!(get_thread_replies(
+            &pool,
+            community,
+            root.id.as_bytes(),
+            Some(10),
+            2,
+            Some(&after)
+        )
+        .await
+        .expect("true EOF")
+        .is_empty());
+    }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn get_thread_replies_bounds_corrupt_row_refills() {
+        let pool = setup_pool().await;
+        let author = Keys::generate();
+        let (channel, community) = create_test_channel(
+            &pool,
+            &format!("corrupt-bound-{}", Uuid::new_v4()),
+            ChannelType::Stream,
+            ChannelVisibility::Open,
+            None,
+            author.public_key().to_bytes().as_slice(),
+            None,
+        )
+        .await
+        .expect("channel");
+        let root = make_stream_event(&author, "root");
+        insert_event_with_thread_metadata(&pool, community, &root, Some(channel.id), None)
+            .await
+            .expect("root");
+        let mut corrupt_ids = Vec::new();
+        for index in 0..65 {
+            let reply = EventBuilder::new(Kind::Custom(9), format!("reply-{index}"))
+                .custom_created_at(nostr::Timestamp::from(
+                    root.created_at.as_secs() + index + 1,
+                ))
+                .sign_with_keys(&author)
+                .expect("reply");
+            insert_event_with_thread_metadata(
+                &pool,
+                community,
+                &reply,
+                Some(channel.id),
+                Some(ThreadMetadataParams {
+                    event_id: reply.id.as_bytes(),
+                    event_created_at: event_created_at(&reply),
+                    channel_id: channel.id,
+                    parent_event_id: Some(root.id.as_bytes()),
+                    parent_event_created_at: Some(event_created_at(&root)),
+                    root_event_id: Some(root.id.as_bytes()),
+                    root_event_created_at: Some(event_created_at(&root)),
+                    depth: 1,
+                    broadcast: false,
+                }),
+            )
+            .await
+            .expect("reply metadata");
+            if index < 64 {
+                corrupt_ids.push(reply.id.as_bytes().to_vec());
+            }
+        }
+        sqlx::query("UPDATE events SET sig = $1 WHERE community_id = $2 AND id = ANY($3)")
+            .bind(vec![0u8; 32])
+            .bind(community.as_uuid())
+            .bind(&corrupt_ids)
+            .execute(&pool)
+            .await
+            .expect("damage first 64 rows");
+        let error = get_thread_replies(&pool, community, root.id.as_bytes(), Some(10), 1, None)
+            .await
+            .expect_err("refill cap must fail, never return partial EOF or scan to the tail");
+        assert!(
+            matches!(error, crate::error::DbError::InvalidData(ref message)
+            if message.contains("corrupt-row refill limit"))
+        );
+        // A larger page needs fewer than 64 raw reads and still finds the tail.
+        let replies = get_thread_replies(&pool, community, root.id.as_bytes(), Some(10), 2, None)
+            .await
+            .expect("bounded skip remains supported");
+        assert_eq!(replies.len(), 1);
+        assert_eq!(replies[0].stored_event.event.content, "reply-64");
+        // The final permitted raw page may be valid; accept it at the cap.
+        sqlx::query("UPDATE events SET deleted_at = NOW() WHERE community_id = $1 AND id = $2")
+            .bind(community.as_uuid())
+            .bind(&corrupt_ids[0])
+            .execute(&pool)
+            .await
+            .expect("hide one corrupt row");
+        let boundary = get_thread_replies(&pool, community, root.id.as_bytes(), Some(10), 1, None)
+            .await
+            .expect("valid final refill page must succeed");
+        assert_eq!(boundary[0].stored_event.event.content, "reply-64");
     }
 
     /// Insert one top-level event (root metadata, broadcast) into a channel.

crates/buzz-relay/src/api/bridge.rs

@@ -21,6 +21,8 @@ use crate::state::AppState;
 
 use super::{api_error, internal_error, not_found};
 
+mod thread_roots;
+
 pub(crate) async fn enforce_http_admission(
     state: &AppState,
     tenant: &TenantContext,
@@ -1185,10 +1187,30 @@ async fn query_events_authed(
     let mut events: Vec<Value> = Vec::new();
     let mut handled: std::collections::HashSet<usize> = std::collections::HashSet::new();
 
+    let ownership_targets: usize = raw_filters
+        .iter()
+        .zip(&filters)
+        .filter(|(raw, _)| extension_flag(raw, "resolve_thread_roots"))
+        .map(|(_, filter)| filter.ids.as_ref().map_or(0, |ids| ids.len()))
+        .sum();
+    if ownership_targets > 100 {
+        return Err(api_error(
+            StatusCode::BAD_REQUEST,
+            "resolve_thread_roots permits at most 100 targets per request",
+        ));
+    }
+    // Resolve reply owners from retained thread metadata, including tombstones.
+    for (idx, (raw, filter)) in raw_filters.iter().zip(filters.iter()).enumerate() {
+        if extension_flag(raw, "resolve_thread_roots") {
+            events.extend(thread_roots::query(state, tenant, filter, &accessible_channels).await?);
+            handled.insert(idx);
+        }
+    }
+
     // Channel-window filters (`top_level: true`) — the GUI read-model surface.
     // Dispatched first: a window filter is never a feed/thread/catchall query.
     for (idx, (raw, filter)) in raw_filters.iter().zip(filters.iter()).enumerate() {
-        if !extension_flag(raw, "top_level") {
+        if handled.contains(&idx) || !extension_flag(raw, "top_level") {
             continue;
         }
         handle_channel_window_filter(

crates/buzz-relay/src/api/bridge/thread_roots.rs

@@ -0,0 +1,123 @@
+//! Metadata-only recovery after a reply deletion whose live recount was missed.
+
+use axum::{http::StatusCode, Json};
+use buzz_core::{kind::KIND_THREAD_SUMMARY, TenantContext};
+use nostr::{EventBuilder, Filter, Kind, Tag};
+use serde_json::Value;
+use uuid::Uuid;
+
+use super::{api_error, extract_channel_from_filter, internal_error};
+use crate::state::AppState;
+
+type QueryError = (StatusCode, Json<Value>);
+
+fn request_channel(filter: &Filter) -> Result<Uuid, QueryError> {
+    let channel = extract_channel_from_filter(filter).ok_or_else(|| {
+        api_error(
+            StatusCode::BAD_REQUEST,
+            "resolve_thread_roots requires exactly one #h channel",
+        )
+    })?;
+    if !filter.kinds.as_ref().is_some_and(|kinds| {
+        kinds.len() == 1 && kinds.contains(&Kind::Custom(KIND_THREAD_SUMMARY as u16))
+    }) {
+        return Err(api_error(
+            StatusCode::BAD_REQUEST,
+            "resolve_thread_roots requires kinds [39005]",
+        ));
+    }
+    if !filter
+        .ids
+        .as_ref()
+        .is_some_and(|ids| !ids.is_empty() && ids.len() <= 100)
+    {
+        return Err(api_error(
+            StatusCode::BAD_REQUEST,
+            "resolve_thread_roots requires 1 to 100 target ids",
+        ));
+    }
+    Ok(channel)
+}
+
+/// Return signed root summaries from retained, channel-scoped reply metadata.
+pub(super) async fn query(
+    state: &AppState,
+    tenant: &TenantContext,
+    filter: &Filter,
+    accessible_channels: &[Uuid],
+) -> Result<Vec<Value>, QueryError> {
+    let channel = request_channel(filter)?;
+    if !accessible_channels.contains(&channel) {
+        return Ok(Vec::new());
+    }
+    let ids = filter
+        .ids
+        .iter()
+        .flatten()
+        .map(|id| id.as_bytes().to_vec())
+        .collect::<Vec<_>>();
+    let summaries = state
+        .db
+        .resolve_thread_root_summaries(tenant.community(), channel, &ids)
+        .await
+        .map_err(|e| internal_error(&format!("thread ownership summaries: {e}")))?;
+    let mut events = Vec::with_capacity(summaries.len());
+    for (root, summary) in summaries {
+        let root_hex = hex::encode(root);
+        let channel_hex = channel.to_string();
+        let tags = [
+            ["e", root_hex.as_str()],
+            ["d", root_hex.as_str()],
+            ["h", channel_hex.as_str()],
+        ]
+        .into_iter()
+        .map(Tag::parse)
+        .collect::<Result<Vec<_>, _>>()
+        .map_err(|e| internal_error(&format!("thread summary tags: {e}")))?;
+        let content = serde_json::json!({
+            "reply_count": summary.reply_count,
+            "descendant_count": summary.descendant_count,
+            "last_reply_at": summary.last_reply_at.map(|t| t.timestamp()),
+            "participants": summary.participants.iter().map(hex::encode).collect::<Vec<_>>(),
+        });
+        let event = EventBuilder::new(
+            Kind::Custom(KIND_THREAD_SUMMARY as u16),
+            content.to_string(),
+        )
+        .tags(tags)
+        .sign_with_keys(&state.relay_keypair)
+        .map_err(|e| internal_error(&format!("thread summary signing: {e}")))?;
+        events.push(
+            serde_json::to_value(event)
+                .map_err(|e| internal_error(&format!("thread summary serialization: {e}")))?,
+        );
+    }
+    Ok(events)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn filter(ids: usize) -> Filter {
+        serde_json::from_value(serde_json::json!({
+            "kinds": [39005], "#h": ["11111111-1111-4111-8111-111111111111"],
+            "ids": (0..ids).map(|id| format!("{id:064x}")).collect::<Vec<_>>()
+        }))
+        .expect("valid filter")
+    }
+
+    #[test]
+    fn ownership_request_requires_bounded_ids_and_explicit_scope() {
+        assert!(request_channel(&filter(1)).is_ok());
+        assert!(request_channel(&filter(100)).is_ok());
+        assert!(request_channel(&filter(0)).is_err());
+        assert!(request_channel(&filter(101)).is_err());
+        let mut wrong_kind = filter(1);
+        wrong_kind.kinds = Some([Kind::TextNote].into_iter().collect());
+        assert!(request_channel(&wrong_kind).is_err());
+        let mut unscoped = filter(1);
+        unscoped.generic_tags.clear();
+        assert!(request_channel(&unscoped).is_err());
+    }
+}

crates/buzz-test-client/tests/e2e_thread_roots.rs

@@ -0,0 +1,148 @@
+//! Tombstone-aware ownership recovery through the authenticated Nostr bridge.
+//! Run against a local development relay with `--ignored`.
+
+use nostr::{Event, EventBuilder, Keys, Kind, Tag};
+use serde_json::{json, Value};
+
+fn url() -> String {
+    std::env::var("RELAY_URL")
+        .unwrap_or_else(|_| "ws://localhost:3000".into())
+        .replace("ws://", "http://")
+        .replace("wss://", "https://")
+}
+
+async fn post(keys: &Keys, event: &Event) {
+    let response = reqwest::Client::new()
+        .post(format!("{}/events", url()))
+        .header("X-Pubkey", keys.public_key().to_hex())
+        .json(event)
+        .send()
+        .await
+        .expect("event request");
+    assert!(
+        response.status().is_success(),
+        "event status {}",
+        response.status()
+    );
+    let body: Value = response.json().await.expect("event response");
+    assert_eq!(body["accepted"], true, "event rejected: {body}");
+}
+
+async fn query(keys: &Keys, filter: Value) -> Vec<Event> {
+    let response = reqwest::Client::new()
+        .post(format!("{}/query", url()))
+        .header("X-Pubkey", keys.public_key().to_hex())
+        .json(&vec![filter])
+        .send()
+        .await
+        .expect("query request");
+    assert!(
+        response.status().is_success(),
+        "query status {}",
+        response.status()
+    );
+    response.json().await.expect("signed event array")
+}
+
+async fn channel(keys: &Keys, private: bool) -> String {
+    let id = uuid::Uuid::new_v4().to_string();
+    let event = EventBuilder::new(Kind::Custom(9007), "")
+        .tags([
+            Tag::parse(["h", &id]).unwrap(),
+            Tag::parse(["name", &format!("thread-roots-{id}")]).unwrap(),
+            Tag::parse(["channel_type", "stream"]).unwrap(),
+            Tag::parse(["visibility", if private { "private" } else { "open" }]).unwrap(),
+        ])
+        .sign_with_keys(keys)
+        .unwrap();
+    post(keys, &event).await;
+    id
+}
+
+fn request(channel: &str, target: &Event) -> Value {
+    json!({"kinds": [39005], "ids": [target.id.to_hex()], "#h": [channel], "resolve_thread_roots": true})
+}
+
+#[tokio::test]
+#[ignore]
+async fn deleted_reply_roots_are_recovered_without_content_or_scope_leaks() {
+    let owner = Keys::generate();
+    let outsider = Keys::generate();
+    let other_channel = channel(&owner, false).await;
+    let bounded_filter = json!({
+        "kinds": [39005], "#h": [&other_channel], "resolve_thread_roots": true,
+        "ids": (0..60).map(|id| format!("{id:064x}")).collect::<Vec<_>>()
+    });
+    let oversized = reqwest::Client::new()
+        .post(format!("{}/query", url()))
+        .header("X-Pubkey", owner.public_key().to_hex())
+        .json(&vec![bounded_filter.clone(), bounded_filter])
+        .send()
+        .await
+        .expect("bounded batch");
+    assert_eq!(
+        oversized.status(),
+        reqwest::StatusCode::BAD_REQUEST,
+        "the target cap applies across filters, not just within each filter"
+    );
+    for deletion_kind in [5, 9005] {
+        let channel_id = channel(&owner, true).await;
+        let root = EventBuilder::new(Kind::Custom(9), "root")
+            .tags([Tag::parse(["h", &channel_id]).unwrap()])
+            .sign_with_keys(&owner)
+            .unwrap();
+        post(&owner, &root).await;
+        let reply = EventBuilder::new(Kind::Custom(9), "deleted secret payload")
+            .tags([
+                Tag::parse(["h", &channel_id]).unwrap(),
+                Tag::parse(["e", &root.id.to_hex(), "", "reply"]).unwrap(),
+            ])
+            .sign_with_keys(&owner)
+            .unwrap();
+        post(&owner, &reply).await;
+        let deletion = EventBuilder::new(Kind::Custom(deletion_kind), "")
+            .tags([
+                Tag::parse(["h", &channel_id]).unwrap(),
+                Tag::parse(["e", &reply.id.to_hex()]).unwrap(),
+            ])
+            .sign_with_keys(&owner)
+            .unwrap();
+        post(&owner, &deletion).await;
+
+        let ordinary = query(
+            &owner,
+            json!({"kinds":[9],"ids":[reply.id.to_hex()],"#h":[channel_id]}),
+        )
+        .await;
+        assert!(
+            ordinary.is_empty(),
+            "ordinary history must exclude the deleted payload"
+        );
+        let summaries = query(&owner, request(&channel_id, &reply)).await;
+        assert_eq!(summaries.len(), 1, "metadata must survive deletion");
+        let summary = &summaries[0];
+        summary.verify().expect("relay signature");
+        assert_eq!(summary.kind, Kind::Custom(39005));
+        assert!(summary
+            .tags
+            .iter()
+            .any(|tag| tag.as_slice() == ["e", &root.id.to_hex()]));
+        let counts: Value = serde_json::from_str(&summary.content).unwrap();
+        assert_eq!(counts["reply_count"], 0);
+        assert_eq!(counts["descendant_count"], 0);
+        assert!(!summary.content.contains("deleted secret payload"));
+
+        assert!(
+            query(&owner, request(&other_channel, &reply))
+                .await
+                .is_empty(),
+            "wrong channel must not leak ownership"
+        );
+        assert!(
+            query(&outsider, request(&channel_id, &reply))
+                .await
+                .is_empty(),
+            "private channel membership is required"
+        );
+    }
+}

docs/nips/NIP-CW.md

@@ -205,3 +205,25 @@ A client with neither an authenticated transport nor a verifiable relay identity
 - **NIP-50** and relay-side search: sibling precedent — a relay-computed view requested through extended filter fields, invisible to relays that do not implement it.
 - **NIP-98**: Authenticates the HTTP query surface Buzz serves windows on.
 - **NIP-11**: Names the relay identity that signs overlays and the natural place to advertise support.
+
+## Recovering summaries after deletion
+
+The HTTP bridge additionally supports `resolve_thread_roots: true` with
+`kinds: [39005]`, exactly one accessible `#h` channel, and 1–100 full event
+`ids`. The IDs identify target replies, including soft-deleted replies,
+not the IDs of the returned summary events. The request permits at most
+100 targets across all such filters.
+
+The relay resolves ownership from retained thread metadata on the writer,
+checks both target and root channel scope, and returns one signed
+`kind:39005` summary per distinct owning root, including zero counts.
+It returns no original target content, author, or signature. Missing or
+non-reply targets and inaccessible channels produce no summaries.
+These reads use the writer so deletion recovery does not depend on
+replica replay or the delivery of a live summary. No bounds event is returned.
+
+Clients use this bounded metadata operation when a deletion target has
+left their reply cache. Ordinary event queries exclude tombstones and
+cannot perform this recovery. This extension requires a supporting relay;
+clients must not interpret an unsupported/empty response as proof that an
+unknown target's thread is empty.

mobile/lib/features/channels/channel_detail_page.dart

@@ -76,6 +76,7 @@ import '../profile/user_profile_sheet.dart';
 import 'small_avatar.dart';
 import 'sticky_date_header.dart';
 import 'thread_detail_page.dart';
+import 'thread_replies_provider.dart';
 import 'timeline_message.dart';
 
 part 'channel_detail_page/message_list.dart';

mobile/lib/features/channels/channel_detail_page/message_bubble.dart

@@ -3,6 +3,7 @@ part of '../channel_detail_page.dart';
 class _MessageBubble extends HookConsumerWidget {
   final TimelineMessage message;
   final bool showAuthor;
+  final bool hasReplies;
   final Map<String, String> channelNames;
   final String currentChannelId;
   final String? currentPubkey;
@@ -15,6 +16,7 @@ class _MessageBubble extends HookConsumerWidget {
   const _MessageBubble({
     required this.message,
     required this.showAuthor,
+    required this.hasReplies,
     required this.channelNames,
     required this.currentChannelId,
     required this.currentPubkey,
@@ -28,6 +30,19 @@ class _MessageBubble extends HookConsumerWidget {
   @override
   Widget build(BuildContext context, WidgetRef ref) {
     final messageSnapshotKey = useMemoized(GlobalKey.new, const []);
+    final hasLocalReplies = ref.watch(
+      threadLocalRepliesProvider(
+        ThreadRepliesArgs(
+          channelId: currentChannelId,
+          rootId: message.rootId ?? message.id,
+        ),
+      ).select(
+        (replies) => replies.any((reply) {
+          final thread = reply.threadReference;
+          return thread.parentId == message.id || thread.rootId == message.id;
+        }),
+      ),
+    );
     // Watch only this user's profile to avoid rebuilding on unrelated cache changes.
     final pk = message.pubkey.toLowerCase();
     final profile =
@@ -115,9 +130,9 @@ class _MessageBubble extends HookConsumerWidget {
           borderRadius: BorderRadius.circular(Radii.md),
           highlightColor: context.colors.primary.withValues(alpha: 0.1),
           snapshotKey: messageSnapshotKey,
-          // Tap opens the thread; long-press still opens the action sheet.
+          // Tap opens existing threads; long-press can start a new one.
           // MessageContent handles mention, channel-link, and media taps.
-          onTap: allMessages == null
+          onTap: (!hasReplies && !hasLocalReplies) || allMessages == null
               ? null
               : () => Navigator.of(context).push(
                   MaterialPageRoute<void>(

mobile/lib/features/channels/channel_detail_page/message_list.dart

@@ -851,6 +851,7 @@ class _MessageList extends HookConsumerWidget {
                           _MessageBubble(
                             message: message,
                             showAuthor: showAuthor,
+                            hasReplies: entry.summary != null,
                             channelNames: channelNamesMap,
                             currentChannelId: channelId,
                             currentPubkey: currentPubkey,

mobile/lib/features/channels/channel_detail_page/system_rows.dart

@@ -613,8 +613,9 @@ class _ThreadSummaryRow extends ConsumerWidget {
                 TextSpan(
                   children: [
                     TextSpan(
-                      text:
-                          '${summary.replyCount} ${summary.replyCount == 1 ? 'reply' : 'replies'}',
+                      text: summary.isCountPending
+                          ? 'Replies'
+                          : '${summary.replyCount}${summary.isLowerBound ? '+' : ''} ${summary.replyCount == 1 ? 'reply' : 'replies'}',
                       style: replyPreviewTextStyle.copyWith(
                         color: context.colors.primary,
                       ),

mobile/lib/features/channels/channel_messages_provider.dart

@@ -6,6 +6,15 @@ import 'channel_event_order.dart';
 import 'pending_local_messages_provider.dart';
 import 'channel_window.dart';
 import 'thread_replies_provider.dart';
+import 'thread_summary_refresh_queue.dart';
+import 'thread_reply_ownership.dart';
+import 'timeline_message.dart';
+
+part 'channel_messages_provider/thread_summaries.dart';
+
+// Channel history keeps partial reply evidence; full threads stay route-scoped.
+const _maxCachedRepliesPerRoot = 256;
+const _maxCachedReplies = 2048;
 
 const _channelLiveEventKinds = [
   ...EventKind.channelEventKinds,
@@ -28,9 +37,43 @@ class ChannelMessagesNotifier extends Notifier<AsyncValue<List<NostrEvent>>> {
   final Set<String> _liveSummaryRootsDuringInitialWindowQuery = {};
   final Map<String, NostrEvent> _deepLinkEvents = {};
   final Set<String> _retainedDeepLinkEventIds = {};
+  final Map<String, String> _localReplyRoots = {};
+  final Map<String, ChannelWindowThreadSummary> _queryThreadSummaries = {};
+  final Map<String, ChannelWindowThreadSummary> _overflowFloors = {};
+  final Map<String, int> _threadQueryVersions = {};
+  final Map<String, int> _threadEvidenceVersions = {};
+  int _threadQuerySerial = 0;
+  final _deletionSummaryUncertainty = <String, _DeletionSummaryUncertainty>{};
+  final _replyOwnership = ThreadReplyOwnership();
+  final _processedDeletionTargets = <String>{};
+  final _deferredDeletionTargets =
+      <String, ({Set<String> candidates, int version})>{};
+  late final _deletionOwnerQueue = _DeletionOwnerQueue(
+    onCapacity: _drainDeferredDeletionOwners,
+    canRun: () =>
+        ref.mounted &&
+        _hasListeners &&
+        ref.read(relaySessionProvider).status == SessionStatus.connected,
+  );
+  bool _hasListeners = true;
+  late final _summaryRefreshes = ThreadSummaryRefreshQueue(
+    refresh: _refreshOverflowSummary,
+    canRun: () =>
+        ref.mounted &&
+        _hasListeners &&
+        ref.read(relaySessionProvider).status == SessionStatus.connected,
+    onSettled: () {
+      final events = _lastKnownMessages;
+      if (events != null) state = AsyncData(events.toList());
+    },
+  );
 
   ChannelMessagesNotifier(this.channelId);
 
+  bool get _summaryMounted => ref.mounted;
+  RelaySessionNotifier get _summarySession =>
+      ref.read(relaySessionProvider.notifier);
+
   /// Last successfully loaded messages, preserved across reconnections so the
   /// UI can show stale data instead of a blank loading spinner.
   List<NostrEvent>? _lastKnownMessages;
@@ -41,14 +84,48 @@ class ChannelMessagesNotifier extends Notifier<AsyncValue<List<NostrEvent>>> {
   /// empty value returned while the relay is not yet connected.
   bool get hasLoadedMessages => _lastKnownMessages != null;
 
-  Map<String, ChannelWindowThreadSummary> get threadSummaries =>
-      channelWindowThreadSummaries(_windowStore);
+  Map<String, ChannelWindowThreadSummary> get threadSummaries => {
+    for (final entry in _baseThreadSummaries.entries)
+      entry.key: _deletionSummaryUncertainty.containsKey(entry.key)
+          ? ChannelWindowThreadSummary(
+              replyCount: entry.value.replyCount,
+              descendantCount: entry.value.descendantCount,
+              lastReplyAt: entry.value.lastReplyAt,
+              participantPubkeys: entry.value.participantPubkeys,
+              isLowerBound: true,
+              isCountPending: true,
+            )
+          : entry.value,
+  };
+
+  void _publishSummaryChange() {
+    final events = _lastKnownMessages;
+    if (events != null) state = AsyncData(events.toList());
+  }
+
+  Map<String, ChannelWindowThreadSummary> get _baseThreadSummaries => {
+    ...channelWindowThreadSummaries(_windowStore),
+    ..._queryThreadSummaries,
+    ..._overflowFloors,
+  };
 
   @override
   AsyncValue<List<NostrEvent>> build() {
     final sessionState = ref.watch(relaySessionProvider);
+    ref.onCancel(() {
+      _hasListeners = false;
+      _deletionOwnerQueue.pause();
+      _summaryRefreshes.pause();
+    });
+    ref.onResume(() {
+      _hasListeners = true;
+      Future.microtask(_summaryRefreshes.resume);
+      Future.microtask(_deletionOwnerQueue.resume);
+    });
     ref.onDispose(() {
       _initVersion++;
+      _deletionOwnerQueue.pause();
+      _summaryRefreshes.pause();
       _clearSubscription();
     });
 
@@ -62,6 +139,16 @@ class ChannelMessagesNotifier extends Notifier<AsyncValue<List<NostrEvent>>> {
 
     _reachedOldest = false;
     _windowStore = const ChannelWindowStore.empty();
+    // A reconnect window can lag a confirmed thread query and contains only
+    // top-level rows. Keep known replies as summary evidence, including their
+    // deletion markers so a reset cannot resurrect a deleted reply.
+    for (final event in _lastKnownMessages ?? const <NostrEvent>[]) {
+      if (event.threadReference.parentId != null ||
+          event.kind == EventKind.deletion ||
+          event.kind == EventKind.nip29DeleteEvent) {
+        _mergeWindowEventIntoStore(event);
+      }
+    }
     _usingChannelWindow = false;
     _initialWindowQueryInFlight = false;
     _liveSummaryRootsDuringInitialWindowQuery.clear();
@@ -103,6 +190,7 @@ class ChannelMessagesNotifier extends Notifier<AsyncValue<List<NostrEvent>>> {
         );
       }
 
+      final historyVersion = ++_threadQuerySerial;
       final history = await _fetchNewestHistory(session);
       if (!_isCurrentInit(initVersion)) return;
       _confirmLocalMessages(history.map((event) => event.id));
@@ -114,7 +202,10 @@ class ChannelMessagesNotifier extends Notifier<AsyncValue<List<NostrEvent>>> {
         ...history.where((event) => existingIds.add(event.id)),
       ]);
       _lastKnownMessages = merged;
+      _reconcileRetainedSummaryPayloads(merged);
+      _reconcileRetainedAggregates(historyVersion);
       state = AsyncData(merged);
+      _summaryRefreshes.resume();
     } catch (e, st) {
       if (!_isCurrentInit(initVersion)) return;
       final fallbackMessages = state.value ?? _lastKnownMessages;
@@ -129,6 +220,7 @@ class ChannelMessagesNotifier extends Notifier<AsyncValue<List<NostrEvent>>> {
     } finally {
       if (_isCurrentInit(initVersion)) {
         _initInFlight = false;
+        _deletionOwnerQueue.resume();
       }
     }
   }
@@ -138,6 +230,7 @@ class ChannelMessagesNotifier extends Notifier<AsyncValue<List<NostrEvent>>> {
   ) async {
     try {
       _initialWindowQueryInFlight = true;
+      final pageVersion = ++_threadQuerySerial;
       final page = await _fetchWindowPage(session, null);
       _initialWindowQueryInFlight = false;
       _windowStore = replaceNewestChannelWindow(
@@ -146,7 +239,9 @@ class ChannelMessagesNotifier extends Notifier<AsyncValue<List<NostrEvent>>> {
         retainLiveSummaryRootIds: _liveSummaryRootsDuringInitialWindowQuery,
       );
       _liveSummaryRootsDuringInitialWindowQuery.clear();
+      _pruneOffWindowReplies();
       _usingChannelWindow = true;
+      _reconcilePageSummaries(page, pageVersion);
       _reachedOldest = !channelWindowHasMore(_windowStore);
       return flattenChannelWindowEvents(_windowStore);
     } catch (error) {
@@ -187,7 +282,13 @@ class ChannelMessagesNotifier extends Notifier<AsyncValue<List<NostrEvent>>> {
     },
   );
 
-  void _handleLiveEvent(NostrEvent event, {bool authoritative = true}) {
+  void _handleLiveEvent(
+    NostrEvent event, {
+    bool authoritative = true,
+    int? summaryVersion,
+  }) {
+    final before = _lastKnownMessages ?? const <NostrEvent>[];
+    _replyOwnership.record([event]);
     // Invalidate the thread query independently of the selected channel-history
     // path. The websocket fallback does not merge through the window store.
     _invalidateThreadReplies(event);
@@ -198,7 +299,7 @@ class ChannelMessagesNotifier extends Notifier<AsyncValue<List<NostrEvent>>> {
       final rootId = _initialWindowQueryInFlight
           ? event.getTagValue('e')
           : null;
-      if (_mergeWindowEventIntoStore(event)) {
+      if (_mergeWindowEventIntoStore(event, summaryVersion: summaryVersion)) {
         if (rootId != null) {
           _liveSummaryRootsDuringInitialWindowQuery.add(rootId);
         }
@@ -211,24 +312,56 @@ class ChannelMessagesNotifier extends Notifier<AsyncValue<List<NostrEvent>>> {
       return;
     }
 
-    // Reply ownership and its thread-local overlay must transition together.
-    // The authoritative thread query performs both confirmations after it
-    // contains the reply; a live echo only triggers that query below.
-    if (authoritative && event.threadReference.parentId == null) {
-      _confirmLocalMessages([event.id]);
-    }
     if (_usingChannelWindow) {
-      _handleWindowLiveEvent(event);
+      _handleWindowLiveEvent(event, summaryVersion: summaryVersion);
     } else {
       final current = state.value ?? _lastKnownMessages ?? const <NostrEvent>[];
-      final merged = _mergeEvent(current, event);
+      final merged = _boundEventReplies(_mergeEvent(current, event));
       _lastKnownMessages = merged;
       state = AsyncData(merged);
     }
+    final rootId = event.threadReference.rootId;
+    if (event.threadReference.parentId != null &&
+        rootId != null &&
+        cachedThreadReplyIds(rootId).length >= _maxCachedRepliesPerRoot) {
+      _queueOverflowSummary(rootId);
+    }
+    if (authoritative) {
+      if (event.kind == EventKind.deletion ||
+          event.kind == EventKind.nip29DeleteEvent) {
+        final targets = event.tags
+            .where((tag) => tag.length > 1 && tag[0] == 'e')
+            .map((tag) => tag[1])
+            .toSet();
+        _refreshDeletedSummaries(targets, before);
+      }
+      if (event.kind == EventKind.deletion ||
+          event.kind == EventKind.nip29DeleteEvent) {
+        _confirmIndexedLocalReplies(
+          event.tags
+              .where((tag) => tag.length > 1 && tag[0] == 'e')
+              .map((tag) => tag[1]),
+        );
+      }
+      _localReplyRoots.remove(event.id);
+      // Store the reply first so clearing its overlay cannot disable its root.
+      _confirmLocalMessages([event.id]);
+      final thread = event.threadReference;
+      if (thread.parentId != null && thread.rootId != null) {
+        final localReplies = threadLocalRepliesProvider(
+          ThreadRepliesArgs(channelId: channelId, rootId: thread.rootId!),
+        );
+        if (ref.exists(localReplies)) {
+          ref.read(localReplies.notifier).confirm({event.id});
+        }
+      }
+    }
   }
 
-  void _handleWindowLiveEvent(NostrEvent event) {
-    if (!_mergeWindowEventIntoStore(event)) return;
+  void _handleWindowLiveEvent(NostrEvent event, {int? summaryVersion}) {
+    if (!_mergeWindowEventIntoStore(event, summaryVersion: summaryVersion)) {
+      return;
+    }
     final windowEvents = flattenChannelWindowEvents(_windowStore);
     // Flattening already orders the window. Only merge and sort again when
     // there are retained deep-link events to include.
@@ -262,7 +395,7 @@ class ChannelMessagesNotifier extends Notifier<AsyncValue<List<NostrEvent>>> {
     }
   }
 
-  bool _mergeWindowEventIntoStore(NostrEvent event) {
+  bool _mergeWindowEventIntoStore(NostrEvent event, {int? summaryVersion}) {
     final isTimelineRow = EventKind.channelTimelineContentKinds.contains(
       event.kind,
     );
@@ -288,18 +421,333 @@ class ChannelMessagesNotifier extends Notifier<AsyncValue<List<NostrEvent>>> {
       _windowStore,
       event,
       isTimelineRow: isTimelineRow,
+      // Deep-link roots stay visible independently of the newest window.
+      // Keep their reply evidence under the same payload bounds as other roots.
+      retainOutsideWindow:
+          thread?.parentId != null &&
+          _retainedDeepLinkEventIds.contains(thread?.rootId),
     );
     if (identical(next, _windowStore)) return false;
     _windowStore = next;
+    if (event.kind == EventKind.channelThreadSummary) {
+      _applyLiveThreadSummary(event, summaryVersion);
+    }
+    _trimReplyOverlay();
+    if (event.kind == EventKind.channelThreadSummary) {
+      _reconcileLiveSummaryPayloads(event);
+    }
     return true;
   }
 
+  List<NostrEvent> _boundedReplies(Iterable<NostrEvent> events) {
+    final sorted = events.toList()
+      ..sort((a, b) => compareThreadRepliesChronologically(b, a));
+    final counts = <String?, int>{};
+    final seen = <String>{};
+    final retained = <NostrEvent>[];
+    for (final event in sorted) {
+      if (!seen.add(event.id)) continue;
+      final root = event.threadReference.rootId;
+      final count = counts[root] ?? 0;
+      if (count >= _maxCachedRepliesPerRoot) continue;
+      retained.add(event);
+      counts[root] = count + 1;
+      if (retained.length == _maxCachedReplies) break;
+    }
+    return retained;
+  }
+
+  List<NostrEvent> _boundEventReplies(List<NostrEvent> events) {
+    final replies = events.where(
+      (event) =>
+          EventKind.channelTimelineContentKinds.contains(event.kind) &&
+          event.threadReference.parentId != null,
+    );
+    final retained = _boundedReplies(replies).map((event) => event.id).toSet();
+    _reconcileEvictedReplies(replies, retained);
+    return events
+        .where(
+          (event) =>
+              !EventKind.channelTimelineContentKinds.contains(event.kind) ||
+              event.threadReference.parentId == null ||
+              retained.contains(event.id),
+        )
+        .toList();
+  }
+
+  void _trimReplyOverlay() {
+    final replies = _windowStore.liveOverlay
+        .where((event) => event.threadReference.parentId != null)
+        .toList();
+    if (replies.length <= _maxCachedRepliesPerRoot) return;
+    final retained = _boundedReplies(replies).map((event) => event.id).toSet();
+    if (retained.length == replies.length) return;
+    _reconcileEvictedReplies(replies, retained);
+    _windowStore = ChannelWindowStore(
+      pages: _windowStore.pages,
+      liveOverlay: _windowStore.liveOverlay
+          .where(
+            (event) =>
+                event.threadReference.parentId == null ||
+                retained.contains(event.id),
+          )
+          .toList(),
+      liveAux: _windowStore.liveAux,
+      liveThreadSummaries: _windowStore.liveThreadSummaries,
+    );
+  }
+
+  void _confirmIndexedLocalReplies(Iterable<String> ids) {
+    final targets = ids.toSet();
+    for (final id in targets) {
+      final root = _localReplyRoots.remove(id);
+      if (root == null) continue;
+      final provider = threadLocalRepliesProvider(
+        ThreadRepliesArgs(channelId: channelId, rootId: root),
+      );
+      if (ref.exists(provider)) ref.read(provider.notifier).confirm({id});
+    }
+    _confirmLocalMessages(targets);
+  }
+
   void _confirmLocalMessages(Iterable<String> eventIds) {
     ref
         .read(pendingLocalMessagesProvider(channelId).notifier)
         .confirm(eventIds);
   }
 
+  /// Reserves a version for a thread query without treating it as evidence.
+  int beginThreadQuery(String rootId) => _reserveThreadQuery(rootId);
+
+  /// Releases a failed or disposed query while preserving newer evidence.
+  void failThreadQuery(String rootId, int version) =>
+      _retireThreadQuery(rootId, version);
+
+  /// Snapshots unacknowledged replies that still require explicit deletion proof.
+  Set<String> unconfirmedThreadReplyIds(String rootId) => {
+    for (final entry in _localReplyRoots.entries)
+      if (entry.value == rootId) entry.key,
+  };
+
+  /// Reserves the next bounded proof batch, rotating pending sends fairly.
+  List<String> nextThreadDeletionProofTargets(Set<String> missingIds) =>
+      _rotateDeletionProofTargets(missingIds);
+
+  /// Snapshots cached overlay replies covered by a query for this outer root.
+  Set<String> cachedThreadReplyIds(String rootId) {
+    final deletedIds = {
+      for (final event in [
+        ..._windowStore.liveAux,
+        if (!_usingChannelWindow) ...?_lastKnownMessages,
+        for (final page in _windowStore.pages) ...page.aux,
+      ])
+        if (event.kind == EventKind.deletion ||
+            event.kind == EventKind.nip29DeleteEvent)
+          for (final tag in event.tags)
+            if (tag.length > 1 && tag[0] == 'e') tag[1],
+    };
+    return {
+      for (final entry in _localReplyRoots.entries)
+        if (entry.value == rootId && !deletedIds.contains(entry.key)) entry.key,
+      for (final event in _cachedThreadEvents)
+        if (event.threadReference.parentId != null &&
+            event.threadReference.rootId == rootId &&
+            !deletedIds.contains(event.id))
+          event.id,
+    };
+  }
+
+  /// Applies explicit deletion evidence fetched for cached replies.
+  /// [scopedTargetIds] are targets of the authenticated channel-scoped query;
+  /// a matching target permits kind-5 wire events without an h tag, including
+  /// multi-target markers that also name events outside this query.
+  /// [reconciledTargetIds] were already excluded by an applied complete scan.
+  void cacheThreadDeletions(
+    Iterable<NostrEvent> deletions, {
+    Set<String> scopedTargetIds = const {},
+    Set<String> reconciledTargetIds = const {},
+  }) {
+    var events = state.value ?? _lastKnownMessages ?? const <NostrEvent>[];
+    for (final event in deletions) {
+      final targets = event.tags
+          .where((tag) => tag.length > 1 && tag[0] == 'e')
+          .map((tag) => tag[1])
+          .toSet();
+      final scopedStandardDeletion =
+          event.kind == EventKind.deletion &&
+          event.channelId == null &&
+          targets.isNotEmpty &&
+          targets.any(scopedTargetIds.contains);
+      if ((!scopedStandardDeletion && event.channelId != channelId) ||
+          (event.kind != EventKind.deletion &&
+              event.kind != EventKind.nip29DeleteEvent)) {
+        throw StateError('Expected a deletion in channel $channelId.');
+      }
+      _replyOwnership.record(events);
+      final affectedTargets = scopedStandardDeletion
+          ? targets
+                .where(
+                  (id) =>
+                      scopedTargetIds.contains(id) ||
+                      _replyOwnership.rootFor(id) != null ||
+                      events.any((cached) => cached.id == id),
+                )
+                .toSet()
+          : targets;
+      final reconciled = affectedTargets.intersection(reconciledTargetIds);
+      _retireDeferredDeletionTargets(reconciled);
+      _rememberDeletionTargets(reconciled);
+      final before = events;
+      _mergeWindowEventIntoStore(event);
+      events = _mergeEvent(events, event);
+      _refreshDeletedSummaries(affectedTargets, before);
+      _confirmIndexedLocalReplies(affectedTargets);
+    }
+    _lastKnownMessages = events;
+    state = AsyncData(events);
+  }
+
+  void _pruneOffWindowReplies() {
+    _rememberDeletionTargets({
+      for (final event in [
+        ..._windowStore.liveAux,
+        for (final page in _windowStore.pages) ...page.aux,
+      ])
+        if (event.kind == EventKind.deletion ||
+            event.kind == EventKind.nip29DeleteEvent)
+          for (final tag in event.tags)
+            if (tag.length > 1 && tag[0] == 'e') tag[1],
+    });
+    final roots = {
+      for (final page in _windowStore.pages)
+        for (final row in page.rows) row.event.id,
+      for (final event in _windowStore.liveOverlay)
+        if (event.threadReference.parentId == null) event.id,
+      ..._retainedDeepLinkEventIds,
+    };
+    bool keepReply(NostrEvent event) =>
+        event.threadReference.parentId == null ||
+        roots.contains(event.threadReference.rootId);
+    final overlay = _windowStore.liveOverlay.where(keepReply).toList();
+    final retainedIds = {...roots, ...overlay.map((event) => event.id)};
+    _windowStore = ChannelWindowStore(
+      pages: _windowStore.pages,
+      liveOverlay: overlay,
+      liveAux: _windowStore.liveAux
+          .where(
+            (event) => event.tags.any(
+              (tag) =>
+                  tag.length > 1 &&
+                  tag[0] == 'e' &&
+                  retainedIds.contains(tag[1]),
+            ),
+          )
+          .toList(),
+      liveThreadSummaries: _windowStore.liveThreadSummaries,
+    );
+    // Keep the currently displayed snapshot intact until the next window
+    // publication drops off-window roots and their reply evidence together.
+  }
+
+  /// Publishes an insertion-complete thread scan, preserving aggregate facts
+  /// before payload eviction. Only absent pre-query, accepted IDs are removed;
+  /// in-flight sends and arrivals after the query began remain provisional.
+  /// Returns whether this scan applied, rather than losing to newer evidence.
+  bool cacheCompleteThreadQuery(
+    String rootId,
+    Set<String> queriedIds,
+    List<NostrEvent> replies, {
+    Set<String> provisionalReplyIds = const {},
+    int? queryVersion,
+  }) {
+    if (queryVersion != null && _threadQueryVersions[rootId] != queryVersion) {
+      return false;
+    }
+    final evidenceVersion = queryVersion ?? ++_threadQuerySerial;
+    _setThreadQueryVersion(rootId, evidenceVersion);
+    _overflowFloors.remove(rootId);
+    _clearDeletionUncertainty(rootId, evidenceVersion);
+    final resultIds = replies.map((event) => event.id).toSet();
+    final missing = queriedIds.difference(resultIds)
+      ..removeAll(_localReplyRoots.keys)
+      ..removeAll(provisionalReplyIds);
+    _rememberDeletionTargets(missing);
+    _retireDeferredDeletionTargets(missing);
+    final retainedIds = cachedThreadReplyIds(rootId);
+    final later = _cachedThreadEvents
+        .where(
+          (event) =>
+              event.threadReference.rootId == rootId &&
+              retainedIds.contains(event.id) &&
+              (!queriedIds.contains(event.id) ||
+                  provisionalReplyIds.contains(event.id)) &&
+              !resultIds.contains(event.id),
+        )
+        .toList();
+    _windowStore = ChannelWindowStore(
+      pages: _windowStore.pages,
+      liveOverlay: _windowStore.liveOverlay
+          .where((event) => !missing.contains(event.id))
+          .toList(),
+      liveAux: _windowStore.liveAux,
+      liveThreadSummaries: _windowStore.liveThreadSummaries,
+    );
+    final current = state.value ?? _lastKnownMessages ?? const <NostrEvent>[];
+    final remaining = current
+        .where((event) => !missing.contains(event.id))
+        .toList();
+    _lastKnownMessages = remaining;
+    state = AsyncData(remaining);
+    final aggregate = {
+      ...{for (final event in replies) event.id: event},
+      ...{for (final event in later) event.id: event},
+    }.values.toList()..sort(compareThreadRepliesChronologically);
+    _queryThreadSummaries.remove(rootId);
+    _queryThreadSummaries[rootId] = ChannelWindowThreadSummary(
+      replyCount: aggregate
+          .where((event) => event.threadReference.parentId == rootId)
+          .length,
+      descendantCount: aggregate.length,
+      lastReplyAt: aggregate.isEmpty ? null : aggregate.last.createdAt,
+      participantPubkeys: aggregate.reversed
+          .map((event) => event.pubkey)
+          .toSet()
+          .take(5)
+          .toList(),
+    );
+    _reconcileNestedThreadSummaries(rootId, aggregate, evidenceVersion);
+    // Metadata is small but still bounded independently of payload retention.
+    while (_queryThreadSummaries.length > 2048) {
+      _queryThreadSummaries.remove(_queryThreadSummaries.keys.first);
+    }
+    cacheConfirmedThreadReplies(replies);
+    return true;
+  }
+
+  /// Caches confirmed thread replies before their optimistic overlay is cleared.
+  /// Thread queries must not invalidate themselves as live relay events do.
+  void cacheConfirmedThreadReplies(Iterable<NostrEvent> replies) {
+    _replyOwnership.record(replies);
+    var events = state.value ?? _lastKnownMessages ?? const <NostrEvent>[];
+    for (final reply in _boundedReplies(replies)) {
+      if (reply.channelId != channelId ||
+          reply.threadReference.parentId == null) {
+        throw StateError('Expected a reply in channel $channelId.');
+      }
+      _mergeWindowEventIntoStore(reply);
+      if (!_usingChannelWindow) {
+        events = _mergeEvent(events, reply);
+      }
+    }
+    if (_usingChannelWindow) {
+      events = _withDeepLinkEvents(flattenChannelWindowEvents(_windowStore));
+    }
+    events = _boundEventReplies(events);
+    _lastKnownMessages = events;
+    state = AsyncData(events);
+    _confirmIndexedLocalReplies(replies.map((event) => event.id));
+  }
+
   /// Adds a just-signed outgoing message before the relay acknowledges it.
   /// The live relay echo is deduplicated by event id.
   void addLocalMessage(NostrEvent event) {
@@ -310,6 +758,7 @@ class ChannelMessagesNotifier extends Notifier<AsyncValue<List<NostrEvent>>> {
       if (rootId == null) {
         throw StateError('Reply ${event.id} has a parent but no thread root.');
       }
+      _localReplyRoots[event.id] = rootId;
       ref
           .read(
             threadLocalRepliesProvider(
@@ -333,11 +782,21 @@ class ChannelMessagesNotifier extends Notifier<AsyncValue<List<NostrEvent>>> {
     _handleLiveEvent(event, authoritative: false);
   }
 
-  /// Releases rollback ownership after the publish future succeeds. The
-  /// optimistic row (and any thread overlay) remains visible until relay data
-  /// replaces it, because OK and EVENT delivery are unordered.
+  /// Releases rollback ownership after the publish future succeeds.
+  /// Accepted replies move into the bounded channel cache even if the relay's
+  /// EVENT echo never arrives, so an unobserved thread overlay can dispose.
   void completeLocalMessage(String eventId) {
-    _confirmLocalMessages([eventId]);
+    final accepted = ref
+        .read(pendingLocalMessagesProvider(channelId).notifier)
+        .take(eventId);
+    if (accepted?.threadReference.parentId != null) {
+      cacheConfirmedThreadReplies([accepted!]);
+      final root = accepted.threadReference.rootId;
+      if (root != null &&
+          cachedThreadReplyIds(root).length >= _maxCachedRepliesPerRoot) {
+        _queueOverflowSummary(root);
+      }
+    }
   }
 
   /// Rolls back a local message when its publish is rejected or times out.
@@ -347,6 +806,7 @@ class ChannelMessagesNotifier extends Notifier<AsyncValue<List<NostrEvent>>> {
         .take(eventId);
     if (pending == null) return;
 
+    _localReplyRoots.remove(eventId);
     final thread = pending.threadReference;
     if (thread.parentId != null) {
       final rootId = thread.rootId;
@@ -437,6 +897,7 @@ class ChannelMessagesNotifier extends Notifier<AsyncValue<List<NostrEvent>>> {
       state.value ?? _lastKnownMessages ?? const [],
     );
     _lastKnownMessages = merged;
+    _reconcileRetainedSummaryPayloads(events);
     state = AsyncData(merged);
   }
 
@@ -467,13 +928,16 @@ class ChannelMessagesNotifier extends Notifier<AsyncValue<List<NostrEvent>>> {
         return false;
       }
       try {
+        final pageVersion = ++_threadQuerySerial;
         final page = await _fetchWindowPage(session, cursor);
         _windowStore = appendOlderChannelWindow(_windowStore, page);
+        _reconcilePageSummaries(page, pageVersion);
         _reachedOldest = !channelWindowHasMore(_windowStore);
         final flattened = _withDeepLinkEvents(
           flattenChannelWindowEvents(_windowStore),
         );
         _lastKnownMessages = flattened;
+        _reconcileRetainedSummaryPayloads(page.rows.map((row) => row.event));
         state = AsyncData(flattened);
         return page.rows.isNotEmpty || page.aux.isNotEmpty;
       } catch (error) {

mobile/lib/features/channels/channel_messages_provider/thread_summaries.dart

@@ -0,0 +1,674 @@
+part of '../channel_messages_provider.dart';
+
+extension _ThreadSummaryState on ChannelMessagesNotifier {
+  void _reconcilePageSummaries(ChannelWindowPage page, int pageVersion) {
+    for (final row in page.rows) {
+      final root = row.event.id;
+      if ((_threadEvidenceVersions[root] ?? 0) > pageVersion) continue;
+      _queryThreadSummaries.remove(root);
+      _overflowFloors.remove(root);
+      _summaryRefreshes.cancel(root);
+      _setThreadQueryVersion(root, pageVersion);
+      _clearDeletionUncertainty(root, pageVersion);
+      if (cachedThreadReplyIds(root).length >
+          (row.thread?.descendantCount ?? 0)) {
+        _queueOverflowSummary(root);
+      }
+    }
+  }
+
+  Iterable<NostrEvent> get _cachedThreadEvents => [
+    ..._windowStore.liveOverlay,
+    if (!_usingChannelWindow) ...?_lastKnownMessages,
+  ];
+
+  List<String> _rotateDeletionProofTargets(Set<String> missingIds) {
+    final targets = missingIds.take(20).toList();
+    // Reuse the pending-send order instead of retaining a second per-root
+    // cursor. It survives route disposal and disappears as sends settle.
+    for (final id in targets) {
+      final root = _localReplyRoots.remove(id);
+      if (root != null) _localReplyRoots[id] = root;
+    }
+    return targets;
+  }
+
+  int _reserveThreadQuery(String root) {
+    final version = ++_threadQuerySerial;
+    _setThreadQueryVersion(root, version, pending: true);
+    return version;
+  }
+
+  void _retireThreadQuery(String root, int version) {
+    if (_threadQueryVersions[root] != version) return;
+    final evidence = _threadEvidenceVersions[root];
+    if (evidence == null) {
+      _threadQueryVersions.remove(root);
+    } else {
+      _threadQueryVersions[root] = evidence;
+    }
+  }
+
+  void _setThreadQueryVersion(
+    String root,
+    int version, {
+    bool pending = false,
+  }) {
+    final current = _threadQueryVersions[root];
+    if (!pending) _threadEvidenceVersions[root] = version;
+    _threadQueryVersions.remove(root);
+    // A page may apply while a newer query is pending. Preserve that query's
+    // reservation so its eventual result can still replace the page evidence.
+    _threadQueryVersions[root] = current != null && current > version
+        ? current
+        : version;
+    while (_threadQueryVersions.length > 2048) {
+      final oldest = _threadQueryVersions.keys.first;
+      _threadQueryVersions.remove(oldest);
+      _threadEvidenceVersions.remove(oldest);
+    }
+  }
+
+  bool _isVisibleRoot(String root) =>
+      _retainedDeepLinkEventIds.contains(root) ||
+      (!_usingChannelWindow &&
+          (_lastKnownMessages?.any((event) => event.id == root) ?? false)) ||
+      _windowStore.pages.any(
+        (page) => page.rows.any((row) => row.event.id == root),
+      ) ||
+      _windowStore.liveOverlay.any((event) => event.id == root);
+
+  bool _hasVisibleThreadRows(String root) =>
+      _isVisibleRoot(root) || _visibleNestedRows(root).isNotEmpty;
+
+  ChannelWindowThreadSummary? _emptyPageSummary(String root) {
+    // A page's implicit zero is no longer the only evidence once replies arrive.
+    if (cachedThreadReplyIds(root).isNotEmpty) return null;
+    for (final page in _windowStore.pages) {
+      for (final row in page.rows) {
+        if (row.event.id != root) continue;
+        if (row.thread != null) return null;
+        return const ChannelWindowThreadSummary(
+          replyCount: 0,
+          descendantCount: 0,
+          lastReplyAt: null,
+          participantPubkeys: [],
+        );
+      }
+    }
+    return null;
+  }
+
+  void _applyLiveThreadSummary(NostrEvent event, int? summaryVersion) {
+    final root = event.getTagValue('e');
+    if (root == null) return;
+    final exact = _queryThreadSummaries[root] ?? _emptyPageSummary(root);
+    if (summaryVersion == null && exact != null) {
+      // WebSocket arrival order cannot establish freshness relative to HTTP
+      // scans. Preserve the completed scan and reconcile conflicting counts.
+      final incoming = parseChannelWindowThreadSummary(event);
+      // Rejected payloads must not regain authority when a later page clears
+      // the query overlay (zero-count page summaries are omitted by the relay).
+      _windowStore = ChannelWindowStore(
+        pages: _windowStore.pages,
+        liveOverlay: _windowStore.liveOverlay,
+        liveAux: _windowStore.liveAux,
+        liveThreadSummaries: {
+          for (final entry in _windowStore.liveThreadSummaries.entries)
+            if (entry.key != root) entry.key: entry.value,
+        },
+      );
+      if (incoming.replyCount != exact.replyCount ||
+          incoming.descendantCount != exact.descendantCount) {
+        _queueOverflowSummary(root, countPending: true);
+      }
+      return;
+    }
+    _queryThreadSummaries.remove(root);
+    _overflowFloors.remove(root);
+    _summaryRefreshes.cancel(root);
+    final version = summaryVersion ?? ++_threadQuerySerial;
+    _setThreadQueryVersion(root, version);
+    _clearDeletionUncertainty(root, version);
+  }
+
+  Iterable<NostrEvent> _visibleNestedRows(String root) =>
+      (_lastKnownMessages ?? const <NostrEvent>[]).where(
+        (event) =>
+            event.threadReference.rootId == root &&
+            event.threadReference.parentId != null &&
+            event.tags.any(
+              (tag) => tag.length > 1 && tag[0] == 'broadcast' && tag[1] == '1',
+            ) &&
+            _isVisibleRoot(event.id),
+      );
+
+  void _reconcileNestedThreadSummaries(
+    String root,
+    List<NostrEvent> replies,
+    int version,
+  ) {
+    final rows = _visibleNestedRows(root).toList();
+    if (rows.isEmpty) return;
+    final entries = buildMainTimelineEntries(
+      formatTimeline([
+        ...{
+          for (final event in [...rows, ...replies]) event.id: event,
+        }.values,
+      ]),
+    );
+    for (final row in rows) {
+      if ((_threadQueryVersions[row.id] ?? 0) > version) continue;
+      final summary = entries
+          .where((entry) => entry.message.id == row.id)
+          .firstOrNull
+          ?.summary;
+      _setThreadQueryVersion(row.id, version);
+      _clearDeletionUncertainty(row.id, version);
+      _overflowFloors.remove(row.id);
+      _queryThreadSummaries.remove(row.id);
+      _queryThreadSummaries[row.id] = ChannelWindowThreadSummary(
+        replyCount: replies
+            .where((event) => event.threadReference.parentId == row.id)
+            .length,
+        descendantCount: summary?.replyCount ?? 0,
+        lastReplyAt: summary?.lastReplyAt,
+        participantPubkeys: summary?.participantPubkeys ?? const [],
+      );
+    }
+  }
+
+  void _reconcileLiveSummaryPayloads(NostrEvent event) {
+    final root = event.getTagValue('e');
+    if (root != null) _reconcileThreadSummaryPayloads(root);
+  }
+
+  void _reconcileRetainedSummaryPayloads(Iterable<NostrEvent> events) {
+    // Evidence can precede the page that first exposes a broadcast branch.
+    final roots = events
+        .map((event) => event.threadReference.rootId)
+        .whereType<String>()
+        .where(_windowStore.liveThreadSummaries.containsKey)
+        .toSet();
+    for (final root in roots) {
+      _reconcileThreadSummaryPayloads(root);
+    }
+  }
+
+  void _reconcileThreadSummaryPayloads(String root) {
+    if (!_hasVisibleThreadRows(root)) return;
+    final summary = _baseThreadSummaries[root];
+    final confirmedIds = cachedThreadReplyIds(root)
+      ..removeAll(_localReplyRoots.keys);
+    // Relay deletion fan-out updates the outer root only. Its visible broadcast
+    // children need the same complete scan to refresh their direct-reply facts.
+    final nestedNeedsRecount =
+        !_queryThreadSummaries.containsKey(root) &&
+        _visibleNestedRows(root).any((row) {
+          final nested = _baseThreadSummaries[row.id];
+          return nested != null &&
+              (nested.replyCount > 0 || nested.descendantCount > 0);
+        });
+    if (!nestedNeedsRecount &&
+        (summary == null || confirmedIds.length <= summary.descendantCount)) {
+      return;
+    }
+    // The summary may lag a new reply, or its deletion marker may have been
+    // missed. A fresh complete query resolves either case without discarding
+    // valid local payloads merely because a summary count is lower.
+    _queueOverflowSummary(root, countPending: true);
+  }
+
+  void _reconcileRetainedAggregates(int historyVersion) {
+    // Bounded history cannot refresh retained roots outside the newest page.
+    // Recount visible cached aggregates, marking their old totals uncertain in
+    // the meantime. Requests or live summaries newer than this history win.
+    final roots = {
+      ..._queryThreadSummaries.keys,
+      ..._overflowFloors.keys,
+      for (final event in _windowStore.liveOverlay)
+        if (event.threadReference.parentId != null &&
+            event.threadReference.rootId != null &&
+            !_localReplyRoots.containsKey(event.id))
+          event.threadReference.rootId!,
+    };
+    for (final root in roots) {
+      if ((_threadQueryVersions[root] ?? 0) > historyVersion) continue;
+      _setThreadQueryVersion(root, historyVersion);
+      if (!_hasVisibleThreadRows(root)) {
+        _queryThreadSummaries.remove(root);
+        _overflowFloors.remove(root);
+        _deletionSummaryUncertainty.remove(root);
+        _summaryRefreshes.cancel(root);
+        continue;
+      }
+      _queueOverflowSummary(root, countPending: true);
+    }
+  }
+
+  void _reconcileEvictedReplies(
+    Iterable<NostrEvent> replies,
+    Set<String> retained,
+  ) {
+    for (final root
+        in replies
+            .where((event) => !retained.contains(event.id))
+            .map((event) => event.threadReference.rootId)
+            .whereType<String>()
+            .toSet()) {
+      if (!_queryThreadSummaries.containsKey(root)) _queueOverflowSummary(root);
+    }
+  }
+
+  void _queueOverflowSummary(String root, {bool countPending = false}) {
+    if (!_hasVisibleThreadRows(root)) return;
+    final ids = cachedThreadReplyIds(root)..removeAll(_localReplyRoots.keys);
+    final events =
+        _cachedThreadEvents.where((event) => ids.contains(event.id)).toList()
+          ..sort(compareThreadRepliesChronologically);
+    final previous = _baseThreadSummaries[root];
+    final count = (previous?.descendantCount ?? 0) > ids.length
+        ? previous!.descendantCount
+        : ids.length;
+    _overflowFloors.remove(root);
+    _overflowFloors[root] = ChannelWindowThreadSummary(
+      replyCount: count,
+      descendantCount: count,
+      lastReplyAt:
+          previous?.lastReplyAt ??
+          (events.isEmpty ? null : events.last.createdAt),
+      participantPubkeys:
+          previous?.participantPubkeys ??
+          events.reversed.map((event) => event.pubkey).toSet().take(5).toList(),
+      isLowerBound: true,
+      isCountPending: countPending || (previous?.isCountPending ?? false),
+    );
+    // The outer scan also owns reconciliation of visible broadcast branches.
+    // Their old totals are uncertain for the entire wait, including exhaustion.
+    for (final row in _visibleNestedRows(root)) {
+      final nested = _baseThreadSummaries[row.id];
+      if (nested == null) continue;
+      _overflowFloors[row.id] = ChannelWindowThreadSummary(
+        replyCount: nested.replyCount,
+        descendantCount: nested.descendantCount,
+        lastReplyAt: nested.lastReplyAt,
+        participantPubkeys: nested.participantPubkeys,
+        isLowerBound: true,
+        isCountPending: true,
+      );
+    }
+    while (_overflowFloors.length > 2048) {
+      _overflowFloors.remove(_overflowFloors.keys.first);
+    }
+    _summaryRefreshes.enqueue(root);
+  }
+
+  Future<void> _refreshOverflowSummary(String root, {int attempt = 0}) async {
+    if (!_hasVisibleThreadRows(root)) return;
+    final generation = _initVersion;
+    final version = beginThreadQuery(root);
+    final snapshot = cachedThreadReplyIds(root);
+    final provisional = unconfirmedThreadReplyIds(root);
+    bool current() =>
+        _summaryMounted &&
+        _hasListeners &&
+        generation == _initVersion &&
+        _threadQueryVersions[root] == version;
+    try {
+      final replies = await fetchCompleteThreadReplies(
+        _summarySession,
+        ThreadRepliesArgs(channelId: channelId, rootId: root),
+        isCurrent: current,
+      );
+      if (current()) {
+        cacheCompleteThreadQuery(
+          root,
+          snapshot,
+          replies,
+          provisionalReplyIds: provisional,
+          queryVersion: version,
+        );
+        if (_summaryRefreshes.isDirty(root)) _queueOverflowSummary(root);
+      }
+    } catch (error) {
+      if (!current()) return;
+      if (attempt < 2) {
+        // Retain this active queue slot during backoff, so retries share the
+        // same concurrency budget and stop when newer work supersedes them.
+        await Future<void>.delayed(Duration(milliseconds: 500 << attempt));
+        if (current()) {
+          await _refreshOverflowSummary(root, attempt: attempt + 1);
+        }
+        return;
+      }
+      debugPrint(
+        '[ChannelMessagesNotifier] thread recount failed for $root: $error',
+      );
+    } finally {
+      _retireThreadQuery(root, version);
+    }
+  }
+
+  void _refreshDeletedSummaries(Set<String> targets, List<NostrEvent> before) {
+    _replyOwnership.record(before);
+    final alreadyDeleted = {
+      ..._processedDeletionTargets,
+      for (final event in before)
+        if (event.kind == EventKind.deletion ||
+            event.kind == EventKind.nip29DeleteEvent)
+          for (final tag in event.tags)
+            if (tag.length > 1 && tag[0] == 'e') tag[1],
+    };
+    final fresh = targets.difference(alreadyDeleted);
+    _rememberDeletionTargets(targets);
+    if (fresh.isEmpty) return;
+    final candidates = _applyDeletedSummaries(fresh, before);
+    final unknown = fresh
+        .where(
+          (id) =>
+              _replyOwnership.rootFor(id) == null &&
+              !before.any((event) => event.id == id),
+        )
+        .toSet();
+    if (unknown.isNotEmpty) _resolveDeletionOwners(unknown, candidates);
+  }
+
+  void _rememberDeletionTargets(Iterable<String> targets) {
+    for (final target in targets) {
+      _processedDeletionTargets.remove(target);
+      _processedDeletionTargets.add(target);
+    }
+    while (_processedDeletionTargets.length > 8192) {
+      _processedDeletionTargets.remove(_processedDeletionTargets.first);
+    }
+  }
+
+  Set<String> _applyDeletedSummaries(
+    Set<String> targets,
+    List<NostrEvent> before,
+  ) {
+    final owners = <String, String>{};
+    for (final id in targets) {
+      final root = _replyOwnership.rootFor(id);
+      if (root != null) owners[id] = root;
+    }
+    // Fence responses started before this deletion, including during debounce.
+    for (final root in owners.values.toSet()) {
+      _setThreadQueryVersion(root, ++_threadQuerySerial);
+    }
+    final summaries = _baseThreadSummaries;
+    final candidates =
+        lowerBoundSummariesAfterDeletion(summaries, before, targets, owners)
+            .entries
+            .where(
+              (entry) =>
+                  entry.value.isCountPending &&
+                  _hasVisibleThreadRows(entry.key),
+            )
+            .map((entry) => entry.key)
+            .toSet();
+    final floors = lowerBoundSummariesAfterDeletion(
+      summaries,
+      before,
+      targets.where(owners.containsKey).toSet(),
+      owners,
+    );
+    for (final entry in floors.entries) {
+      if (!_hasVisibleThreadRows(entry.key)) continue;
+      _overflowFloors[entry.key] = entry.value;
+      // Unknown ownership changes presentation only; never scan every candidate.
+      if (owners.containsValue(entry.key)) _queueOverflowSummary(entry.key);
+    }
+    for (final root in owners.values.toSet()) {
+      // An off-window outer root may have no summary, while its broadcast row
+      // still needs reconciliation after a known child's deletion.
+      if (!floors.containsKey(root)) _queueOverflowSummary(root);
+    }
+    while (_overflowFloors.length > 2048) {
+      _overflowFloors.remove(_overflowFloors.keys.first);
+    }
+    return candidates;
+  }
+
+  void _resolveDeletionOwners(Set<String> targets, Set<String> candidates) {
+    // Register every target before sending the first bounded-queue query. A
+    // successful target must not retire uncertainty from a queued sibling.
+    final versions = {
+      for (final target in targets) target: ++_threadQuerySerial,
+    };
+    for (final root in candidates) {
+      final pending = _deletionSummaryUncertainty[root] ??=
+          _DeletionSummaryUncertainty();
+      for (final version in versions.values) {
+        pending.begin(version);
+      }
+    }
+    while (_deletionSummaryUncertainty.length > 2048) {
+      _deletionSummaryUncertainty.remove(
+        _deletionSummaryUncertainty.keys.first,
+      );
+    }
+    _publishSummaryChange();
+    // The shared queue bounds work across events as well as within each batch.
+    for (final entry in versions.entries) {
+      if (!_enqueueDeletionOwner(entry.key, candidates, entry.value)) {
+        _deferredDeletionTargets[entry.key] = (
+          candidates: candidates,
+          version: entry.value,
+        );
+        while (_deferredDeletionTargets.length > 8192) {
+          final oldest = _deferredDeletionTargets.keys.first;
+          final dropped = _deferredDeletionTargets.remove(oldest)!;
+          _finishDeletionLookup(
+            dropped.candidates,
+            dropped.version,
+            resolved: false,
+          );
+        }
+      }
+    }
+  }
+
+  bool _enqueueDeletionOwner(
+    String target,
+    Set<String> candidates,
+    int version,
+  ) => _deletionOwnerQueue.enqueue(() async {
+    if (!_summaryMounted) return true;
+    return _resolveDeletionOwner(target, candidates, _initVersion, version);
+  });
+
+  void _drainDeferredDeletionOwners() {
+    if (!_summaryMounted) return;
+    while (_deletionOwnerQueue.hasCapacity &&
+        _deferredDeletionTargets.isNotEmpty) {
+      final target = _deferredDeletionTargets.keys.first;
+      final work = _deferredDeletionTargets.remove(target)!;
+      _enqueueDeletionOwner(target, work.candidates, work.version);
+    }
+  }
+
+  void _retireDeferredDeletionTargets(Iterable<String> targets) {
+    for (final target in targets) {
+      final work = _deferredDeletionTargets.remove(target);
+      if (work != null) {
+        _finishDeletionLookup(work.candidates, work.version, resolved: true);
+      }
+    }
+  }
+
+  Future<bool> _resolveDeletionOwner(
+    String target,
+    Set<String> candidates,
+    int generation,
+    int version, {
+    int attempt = 0,
+  }) async {
+    try {
+      final events = await _summarySession.queryRelay([
+        NostrFilter(
+          ids: [target],
+          kinds: const [EventKind.channelThreadSummary],
+          extensions: const {'resolve_thread_roots': true},
+          tags: {
+            '#h': [channelId],
+          },
+          limit: 1,
+        ),
+      ]);
+      if (!_summaryMounted) return true;
+      if (generation != _initVersion) return false;
+      if (events.length > 1) {
+        throw StateError('Expected at most one owner for a deletion target.');
+      }
+      var applied = false;
+      String? resolvedRoot;
+      for (final event in events) {
+        if (event.channelId != channelId ||
+            event.kind != EventKind.channelThreadSummary) {
+          throw StateError('Expected a channel-scoped thread summary.');
+        }
+        final root = event.getTagValue('e');
+        resolvedRoot = root;
+        if (root != null && (_threadQueryVersions[root] ?? 0) <= version) {
+          _handleLiveEvent(event, summaryVersion: version);
+          applied = _threadQueryVersions[root] == version;
+        }
+      }
+      // An empty response does not prove ownership (including on old relays).
+      // A response fenced by a newer query is not applied evidence either:
+      // that query can still fail, leaving the old count uncertain.
+      _finishDeletionLookup(
+        candidates,
+        version,
+        resolved: resolvedRoot != null,
+        unresolvedRoot: applied ? null : resolvedRoot,
+      );
+    } catch (error) {
+      if (!_summaryMounted) return true;
+      if (generation != _initVersion) return false;
+      if (error is! StateError && error is! FormatException && attempt < 2) {
+        // Keep the queue slot and original version while backing off. Retries
+        // cannot exceed the shared concurrency cap or supersede newer evidence.
+        await Future<void>.delayed(Duration(milliseconds: 500 << attempt));
+        if (!_summaryMounted) return true;
+        if (generation != _initVersion) return false;
+        return _resolveDeletionOwner(
+          target,
+          candidates,
+          generation,
+          version,
+          attempt: attempt + 1,
+        );
+      }
+      // Exhausted or invalid responses remain uncertain until fresh evidence.
+      if (_summaryMounted && generation == _initVersion) {
+        _finishDeletionLookup(candidates, version, resolved: false);
+        debugPrint(
+          '[ChannelMessagesNotifier] deletion ownership lookup failed: $error',
+        );
+      }
+    }
+    return true;
+  }
+
+  void _clearDeletionUncertainty(String root, int version) {
+    final pending = _deletionSummaryUncertainty[root];
+    if (pending == null) return;
+    pending.clearThrough(version);
+    if (pending.isEmpty) _deletionSummaryUncertainty.remove(root);
+  }
+
+  void _finishDeletionLookup(
+    Set<String> candidates,
+    int version, {
+    required bool resolved,
+    String? unresolvedRoot,
+  }) {
+    for (final root in candidates) {
+      final pending = _deletionSummaryUncertainty[root];
+      if (pending == null) continue;
+      pending.finish(version, resolved: resolved && root != unresolvedRoot);
+      if (pending.isEmpty) _deletionSummaryUncertainty.remove(root);
+    }
+    _publishSummaryChange();
+  }
+}
+
+// Track uncertainty independently of counts so resolving one deletion cannot
+// erase another pending deletion, or leave unrelated exact counts downgraded.
+class _DeletionSummaryUncertainty {
+  final _requests = <int>{};
+  int? _unresolvedVersion;
+
+  bool get isEmpty => _requests.isEmpty && _unresolvedVersion == null;
+
+  void begin(int version) {
+    _requests.add(version);
+    if (_requests.length > 256) {
+      final oldest = _requests.first;
+      if (oldest > (_unresolvedVersion ?? -1)) _unresolvedVersion = oldest;
+      _requests.remove(oldest);
+    }
+  }
+
+  void finish(int version, {required bool resolved}) {
+    if (!_requests.remove(version)) return;
+    if (!resolved && version > (_unresolvedVersion ?? -1)) {
+      _unresolvedVersion = version;
+    }
+  }
+
+  void clearThrough(int version) {
+    _requests.removeWhere((request) => request <= version);
+    if ((_unresolvedVersion ?? -1) <= version) _unresolvedVersion = null;
+  }
+}
+
+// Bound retained work and concurrent HTTP requests for the whole channel.
+class _DeletionOwnerQueue {
+  final bool Function() canRun;
+  final void Function() onCapacity;
+  final _pending = <Future<bool> Function()>[];
+  int _active = 0;
+  bool _paused = false;
+
+  _DeletionOwnerQueue({required this.canRun, required this.onCapacity});
+
+  bool get hasCapacity => _pending.length + _active < 258;
+
+  bool enqueue(Future<bool> Function() request) {
+    if (!hasCapacity) return false;
+    _pending.add(request);
+    _drain();
+    return true;
+  }
+
+  void pause() => _paused = true;
+
+  void resume() {
+    _paused = false;
+    _drain();
+    onCapacity();
+  }
+
+  void _drain() {
+    while (!_paused && canRun() && _active < 2 && _pending.isNotEmpty) {
+      _active++;
+      _run(_pending.removeAt(0));
+    }
+  }
+
+  Future<void> _run(Future<bool> Function() request) async {
+    try {
+      // Interrupted work keeps its place in the same total admission budget.
+      // Reconnect resumes it with a fresh connection generation, but the
+      // original summary version still fences it against newer root evidence.
+      if (!await request()) _pending.insert(0, request);
+    } finally {
+      _active--;
+      _drain();
+      onCapacity();
+    }
+  }
+}

mobile/lib/features/channels/channel_window.dart

@@ -16,11 +16,19 @@ class ChannelWindowThreadSummary {
   final int? lastReplyAt;
   final List<String> participantPubkeys;
 
+  /// True while capped evidence awaits a complete recount.
+  final bool isLowerBound;
+
+  /// The last known total is awaiting reconciliation and should not be displayed.
+  final bool isCountPending;
+
   const ChannelWindowThreadSummary({
     required this.replyCount,
     required this.descendantCount,
     required this.lastReplyAt,
     required this.participantPubkeys,
+    this.isLowerBound = false,
+    this.isCountPending = false,
   });
 }
 
@@ -273,6 +281,7 @@ ChannelWindowStore mergeLiveChannelWindowEvent(
   ChannelWindowStore current,
   NostrEvent event, {
   required bool isTimelineRow,
+  bool retainOutsideWindow = false,
 }) {
   // A reply doesn't reach the main timeline itself — the root's "N replies" row
   // comes from this summary event, which the relay re-emits on every reply. It
@@ -331,7 +340,8 @@ ChannelWindowStore mergeLiveChannelWindowEvent(
   final oldest = oldestPage?.rows.isEmpty ?? true
       ? null
       : oldestPage!.rows.last.event;
-  if (oldest != null &&
+  if (!retainOutsideWindow &&
+      oldest != null &&
       (event.createdAt < oldest.createdAt ||
           (oldestPage!.hasMore && _compareRelayOrder(event, oldest) >= 0))) {
     return current;

mobile/lib/features/channels/thread_detail_page/nested_thread_summary_row.dart

@@ -76,8 +76,9 @@ class _NestedThreadSummaryRow extends ConsumerWidget {
                 TextSpan(
                   children: [
                     TextSpan(
-                      text:
-                          '${summary.replyCount} ${summary.replyCount == 1 ? 'reply' : 'replies'}',
+                      text: summary.isCountPending
+                          ? 'Replies'
+                          : '${summary.replyCount}${summary.isLowerBound ? '+' : ''} ${summary.replyCount == 1 ? 'reply' : 'replies'}',
                       style: replyPreviewTextStyle.copyWith(
                         color: context.colors.primary,
                       ),

mobile/lib/features/channels/thread_replies_provider.dart

@@ -1,9 +1,11 @@
 import 'dart:async';
 
 import 'package:hooks_riverpod/hooks_riverpod.dart';
+import 'package:hooks_riverpod/misc.dart' show KeepAliveLink;
 
 import '../../shared/relay/relay.dart';
 import 'channel_event_order.dart';
+import 'channel_messages_provider.dart';
 import 'pending_local_messages_provider.dart';
 
 class ThreadRepliesArgs {
@@ -42,20 +44,111 @@ final threadRepliesProvider = FutureProvider.autoDispose
         }
       });
       final session = ref.read(relaySessionProvider.notifier);
-      final replies = <NostrEvent>[];
-      _ThreadCursor? cursor;
-      for (var page = 0; page < 500; page++) {
-        final events = await session.queryRelay([
-          _threadRepliesFilter(args, cursor),
-        ]);
-        replies.addAll(events);
-        if (events.length < 200) return replies;
-        final last = events.last;
-        cursor = _ThreadCursor(createdAt: last.createdAt, eventId: last.id);
+      final channelProvider = channelMessagesProvider(args.channelId);
+      final channelMessages = ref.exists(channelProvider)
+          ? ref.read(channelProvider.notifier)
+          : null;
+      final cachedReplyIds = channelMessages?.cachedThreadReplyIds(args.rootId);
+      final unconfirmedIds = channelMessages?.unconfirmedThreadReplyIds(
+        args.rootId,
+      );
+      final queryVersion = channelMessages?.beginThreadQuery(args.rootId);
+      if (queryVersion != null) {
+        ref.onDispose(
+          () => channelMessages?.failThreadQuery(args.rootId, queryVersion),
+        );
+      }
+      try {
+        final replies = await fetchCompleteThreadReplies(session, args);
+        // Explicit markers also settle unacknowledged local sends, whose
+        // absence cannot prove deletion even in an insertion-complete scan.
+        final missingIds =
+            unconfirmedIds?.difference(
+              replies.map((event) => event.id).toSet(),
+            ) ??
+            <String>{};
+        final deletions = <NostrEvent>[];
+        // One bounded request per scan keeps opening a thread responsive even
+        // when many sends are awaiting ACKs. Excess IDs remain provisional.
+        final targets =
+            channelMessages?.nextThreadDeletionProofTargets(missingIds) ??
+            const <String>[];
+        // Per-target limits keep repeated markers from crowding another ID out.
+        if (targets.isNotEmpty) {
+          deletions.addAll(
+            await session.queryRelay([
+              for (final target in targets)
+                NostrFilter(
+                  kinds: const [EventKind.deletion, EventKind.nip29DeleteEvent],
+                  tags: {
+                    '#h': [args.channelId],
+                    '#e': [target],
+                  },
+                  limit: 1,
+                ),
+            ]),
+          );
+        }
+        if (ref.mounted && ref.exists(channelProvider)) {
+          final channel = ref.read(channelProvider.notifier);
+          final deletedTargets = {
+            for (final event in deletions)
+              for (final tag in event.tags)
+                if (tag.length > 1 && tag[0] == 'e') tag[1],
+          };
+          final applied = channel.cacheCompleteThreadQuery(
+            args.rootId,
+            cachedReplyIds ?? {},
+            replies,
+            provisionalReplyIds: missingIds.difference(deletedTargets),
+            queryVersion: queryVersion,
+          );
+          if (deletions.isNotEmpty) {
+            channel.cacheThreadDeletions(
+              deletions,
+              scopedTargetIds: targets.toSet(),
+              reconciledTargetIds: applied ? missingIds : const {},
+            );
+          }
+        }
+        return replies;
+      } catch (_) {
+        if (queryVersion != null) {
+          channelMessages?.failThreadQuery(args.rootId, queryVersion);
+        }
+        rethrow;
       }
-      throw Exception('Thread ${args.rootId} exceeded the page safety limit.');
     });
 
+/// Exhaustively scans a thread using insertion-complete cursor pages.
+/// [isCurrent] lets a background refresh stop between pages after disposal.
+Future<List<NostrEvent>> fetchCompleteThreadReplies(
+  RelaySessionNotifier session,
+  ThreadRepliesArgs args, {
+  bool Function()? isCurrent,
+}) async {
+  final replies = <NostrEvent>[];
+  // -1 precedes unsigned Nostr timestamps. A non-null cursor selects the
+  // insertion-complete route and writer-verified EOF instead of a stale head.
+  _ThreadCursor? cursor = const _ThreadCursor(
+    createdAt: -1,
+    eventId: '0000000000000000000000000000000000000000000000000000000000000000',
+  );
+  for (var page = 0; page < 500; page++) {
+    if (isCurrent != null && !isCurrent()) {
+      throw StateError('Thread scan superseded');
+    }
+    final events = await session.queryRelay([
+      _threadRepliesFilter(args, cursor),
+    ]);
+    replies.addAll(events);
+    if (events.length < 200) return replies;
+    final last = events.last;
+    cursor = _ThreadCursor(createdAt: last.createdAt, eventId: last.id);
+  }
+  throw Exception('Thread ${args.rootId} exceeded the page safety limit.');
+}
+
 NostrFilter _threadRepliesFilter(
   ThreadRepliesArgs args,
   _ThreadCursor? cursor,
@@ -68,7 +161,8 @@ NostrFilter _threadRepliesFilter(
     },
     limit: 200,
     extensions: {
-      'depth_limit': 64,
+      // The relay binds this as signed i32. Include every representable depth.
+      'depth_limit': 0x7fffffff,
       if (cursor != null) 'thread_cursor': cursor.createdAt,
       if (cursor != null) 'thread_cursor_id': cursor.eventId,
     },
@@ -81,7 +175,19 @@ class ThreadLocalRepliesNotifier extends Notifier<List<NostrEvent>> {
   ThreadLocalRepliesNotifier(this.args);
 
   @override
-  List<NostrEvent> build() => const [];
+  List<NostrEvent> build() {
+    // Keep optimistic replies across route disposal, but release empty overlays.
+    KeepAliveLink? retention;
+    listenSelf((previous, next) {
+      if (next.isNotEmpty) {
+        retention ??= ref.keepAlive();
+      } else {
+        retention?.close();
+        retention = null;
+      }
+    });
+    return const [];
+  }
 
   void add(NostrEvent event) {
     state = _mergeReplies(state, [event]);
@@ -97,12 +203,10 @@ class ThreadLocalRepliesNotifier extends Notifier<List<NostrEvent>> {
   }
 }
 
-final threadLocalRepliesProvider =
-    NotifierProvider.family<
-      ThreadLocalRepliesNotifier,
-      List<NostrEvent>,
-      ThreadRepliesArgs
-    >(ThreadLocalRepliesNotifier.new);
+final threadLocalRepliesProvider = NotifierProvider.autoDispose
+    .family<ThreadLocalRepliesNotifier, List<NostrEvent>, ThreadRepliesArgs>(
+      ThreadLocalRepliesNotifier.new,
+    );
 
 /// Relay-backed replies merged with signed local replies that are still
 /// waiting for acknowledgement.
@@ -123,7 +227,16 @@ final threadRepliesWithLocalProvider = Provider.autoDispose
           final pendingMessagesNotifier = ref.read(
             pendingLocalMessagesProvider(args.channelId).notifier,
           );
+          final channelMessages = ref.read(
+            channelMessagesProvider(args.channelId).notifier,
+          );
+          final confirmedReplies = authoritative
+              .where(
+                (event) => localReplies.any((local) => local.id == event.id),
+              )
+              .toList();
           Future.microtask(() {
+            channelMessages.cacheConfirmedThreadReplies(confirmedReplies);
             localRepliesNotifier.confirm(authoritativeIds);
             pendingMessagesNotifier.confirm(authoritativeIds);
           });

mobile/lib/features/channels/thread_reply_ownership.dart

@@ -0,0 +1,26 @@
+import '../../shared/relay/relay.dart';
+
+/// Bounded identity-only evidence for replies whose payloads may be evicted.
+class ThreadReplyOwnership {
+  final Map<String, String> _roots = {};
+
+  /// Remembers recent ownership without retaining event payloads.
+  void record(Iterable<NostrEvent> events) {
+    for (final event in events) {
+      final thread = event.threadReference;
+      if (!EventKind.channelTimelineContentKinds.contains(event.kind) ||
+          thread.parentId == null ||
+          thread.rootId == null) {
+        continue;
+      }
+      _roots.remove(event.id);
+      _roots[event.id] = thread.rootId!;
+      while (_roots.length > 8192) {
+        _roots.remove(_roots.keys.first);
+      }
+    }
+  }
+
+  /// Returns known ownership or null when the identity evidence was evicted.
+  String? rootFor(String eventId) => _roots[eventId];
+}

mobile/lib/features/channels/thread_summary_refresh_queue.dart

@@ -0,0 +1,143 @@
+import 'dart:async';
+import 'dart:math' as math;
+
+import '../../shared/relay/relay.dart';
+import 'channel_window.dart';
+
+/// Coalesces visible-thread recounts, with at most two concurrent scans.
+class ThreadSummaryRefreshQueue {
+  /// Builds a queue whose owner handles refresh errors and lifecycle validity.
+  ThreadSummaryRefreshQueue({
+    required this.refresh,
+    required this.canRun,
+    required this.onSettled,
+  });
+
+  /// Loads and publishes one root's complete summary.
+  final Future<void> Function(String rootId) refresh;
+
+  /// Whether the owner is connected and still mounted.
+  final bool Function() canRun;
+
+  /// Notifies the owner after an active request releases its slot.
+  final void Function() onSettled;
+  final Set<String> _queued = {};
+  final Set<String> _active = {};
+  Timer? _timer;
+
+  /// Marks a root dirty. Repeated arrivals collapse into one pending scan.
+  void enqueue(String rootId) {
+    _queued.remove(rootId);
+    _queued.add(rootId);
+    // Dropped requests retain their lower-bound display until the thread opens.
+    while (_queued.length > 256) {
+      _queued.remove(_queued.first);
+    }
+    resume();
+  }
+
+  /// Removes queued work when a relay summary already supplied the answer.
+  void cancel(String rootId) => _queued.remove(rootId);
+
+  /// Whether another scan was requested after the active scan began.
+  bool isDirty(String rootId) => _queued.contains(rootId);
+
+  /// Cancels a scheduled wake-up while preserving dirty roots for reconnect.
+  void pause() {
+    _queued.addAll(_active);
+    _timer?.cancel();
+    _timer = null;
+  }
+
+  /// Resumes queued work after the owning connection recovers.
+  void resume() {
+    if (_timer != null || _queued.isEmpty || !canRun()) return;
+    _timer = Timer(const Duration(milliseconds: 200), _drain);
+  }
+
+  void _drain() {
+    _timer = null;
+    if (!canRun()) return;
+    for (final root in _queued.toList()) {
+      if (_active.length >= 2) break;
+      if (_active.contains(root)) continue;
+      _queued.remove(root);
+      _active.add(root);
+      unawaited(_run(root));
+    }
+  }
+
+  Future<void> _run(String root) async {
+    try {
+      await refresh(root);
+    } finally {
+      _active.remove(root);
+      if (canRun()) {
+        onSettled();
+        resume();
+      }
+    }
+  }
+}
+
+/// Conservatively adjusts known totals after explicit deletions, including
+/// targets whose payloads have been evicted. Unknown ownership preserves navigation
+/// and hides the stale number until a complete recount restores exactness.
+Map<String, ChannelWindowThreadSummary> lowerBoundSummariesAfterDeletion(
+  Map<String, ChannelWindowThreadSummary> summaries,
+  List<NostrEvent> events,
+  Set<String> targets,
+  Map<String, String> owners,
+) {
+  final byId = {for (final event in events) event.id: event};
+  final knownRoots = owners.values.toSet();
+  final knownTargetsPerRoot = <String, int>{};
+  for (final root in owners.values) {
+    knownTargetsPerRoot[root] = (knownTargetsPerRoot[root] ?? 0) + 1;
+  }
+  final hasUnknownTarget = targets.any(
+    (id) => !owners.containsKey(id) && !byId.containsKey(id),
+  );
+  final deleted = {
+    ...targets,
+    for (final event in events)
+      if (event.kind == EventKind.deletion ||
+          event.kind == EventKind.nip29DeleteEvent)
+        for (final tag in event.tags)
+          if (tag.length > 1 && tag[0] == 'e') tag[1],
+  };
+  final cached = <String?, int>{};
+  final remaining = <String?, int>{};
+  for (final event in byId.values) {
+    if (!EventKind.channelTimelineContentKinds.contains(event.kind) ||
+        event.threadReference.parentId == null) {
+      continue;
+    }
+    final root = event.threadReference.rootId;
+    cached[root] = (cached[root] ?? 0) + 1;
+    if (!deleted.contains(event.id)) {
+      remaining[root] = (remaining[root] ?? 0) + 1;
+    }
+  }
+  return {
+    for (final entry in summaries.entries)
+      if (knownRoots.contains(entry.key) ||
+          (hasUnknownTarget &&
+              math.max(entry.value.replyCount, entry.value.descendantCount) >
+                  (cached[entry.key] ?? 0)))
+        entry.key: ChannelWindowThreadSummary(
+          replyCount: math.max(
+            0,
+            entry.value.replyCount - (knownTargetsPerRoot[entry.key] ?? 0),
+          ),
+          descendantCount: math.max(
+            remaining[entry.key] ?? 0,
+            entry.value.descendantCount - (knownTargetsPerRoot[entry.key] ?? 0),
+          ),
+          lastReplyAt: entry.value.lastReplyAt,
+          participantPubkeys: entry.value.participantPubkeys,
+          isLowerBound: true,
+          isCountPending: hasUnknownTarget || entry.value.isCountPending,
+        ),
+  };
+}

mobile/lib/features/channels/timeline_message.dart

@@ -219,6 +219,12 @@ class ThreadSummary {
   final String threadHeadId;
   final int replyCount;
 
+  /// Whether this count is a minimum pending a complete recount.
+  final bool isLowerBound;
+
+  /// The last known total is awaiting reconciliation and should not be displayed.
+  final bool isCountPending;
+
   /// Up to 3 most recent unique participant pubkeys.
   final List<String> participantPubkeys;
   final int? lastReplyAt;
@@ -226,6 +232,8 @@ class ThreadSummary {
   const ThreadSummary({
     required this.threadHeadId,
     required this.replyCount,
+    this.isLowerBound = false,
+    this.isCountPending = false,
     required this.participantPubkeys,
     this.lastReplyAt,
   });
@@ -560,10 +568,10 @@ bool _isBroadcastReply(TimelineMessage message) {
 /// order is mobile's own (see [_mergeParticipants]) because this file already
 /// renders relay participants in the order the relay sent them.
 ///
-/// Both halves count descendants, not direct replies: the relay's
-/// `descendant_count` and the locally assembled [_buildDescendantStats]. A badge
-/// on the main timeline stands for the whole thread under that message, so a
-/// reply to a reply has to raise it.
+/// Both halves represent descendants. The relay only increments descendant
+/// counts on the outer root, so a broadcast reply's direct-reply count is also
+/// positive evidence, and a lower bound on its descendants. Locally assembled
+/// [_buildDescendantStats] includes nested replies under every ancestor.
 ThreadSummary? _buildSummary(
   String messageId,
   Map<String, _DescendantStats> descendantStats,
@@ -576,6 +584,8 @@ ThreadSummary? _buildSummary(
 
   return ThreadSummary(
     threadHeadId: messageId,
+    isLowerBound: relay.isLowerBound,
+    isCountPending: relay.isCountPending,
     replyCount: local.replyCount > relay.replyCount
         ? local.replyCount
         : relay.replyCount,
@@ -684,10 +694,19 @@ ThreadSummary? _buildRelaySummary(
   String messageId,
   ChannelWindowThreadSummary? relaySummary,
 ) {
-  if (relaySummary == null || relaySummary.descendantCount <= 0) return null;
+  if (relaySummary == null) return null;
+  // The relay counts direct children on every parent, but descendants only on
+  // the outer root. Broadcast replies can therefore have valid 1/0 counters.
+  final useDirectCount = relaySummary.replyCount > relaySummary.descendantCount;
+  final count = useDirectCount
+      ? relaySummary.replyCount
+      : relaySummary.descendantCount;
+  if (count <= 0) return null;
   return ThreadSummary(
     threadHeadId: messageId,
-    replyCount: relaySummary.descendantCount,
+    replyCount: count,
+    isLowerBound: relaySummary.isLowerBound || useDirectCount,
+    isCountPending: relaySummary.isCountPending,
     participantPubkeys: relaySummary.participantPubkeys.take(3).toList(),
     lastReplyAt: relaySummary.lastReplyAt,
   );

mobile/test/features/channels/channel_detail_page_test.dart

@@ -19,6 +19,7 @@ import 'package:buzz/features/channels/channel.dart';
 import 'package:buzz/features/channels/channel_detail_page.dart';
 import 'package:buzz/features/channels/channel_management_provider.dart';
 import 'package:buzz/features/channels/channel_messages_provider.dart';
+import 'package:buzz/features/channels/channel_window.dart';
 import 'package:buzz/features/channels/channel_mutes/channel_mutes_provider.dart';
 import 'package:buzz/features/channels/channel_mutes/channel_mutes_storage.dart';
 import 'package:buzz/features/channels/channel_stars/channel_stars_provider.dart';
@@ -558,14 +559,25 @@ void main() {
           pubkey: 'bot',
           content: 'Bot message',
         );
+        final reply = _textMsg(
+          id: 'bot-reply',
+          pubkey: 'alice',
+          content: 'Reply to bot',
+          createdAt: 1100,
+          extraTags: const [
+            ['e', 'bot-message', '', 'reply'],
+          ],
+        );
         await tester.pumpWidget(
           _buildTestable(
-            messages: [message],
+            messages: [message, reply],
             users: const {
               'bot': UserProfile(pubkey: 'bot', displayName: 'Bot'),
             },
             loadChannelBotPubkeys: () async => const {'bot'},
-            threadReplies: const {'bot-message': []},
+            threadReplies: {
+              'bot-message': [reply],
+            },
           ),
         );
         await tester.pumpAndSettle();
@@ -10255,6 +10267,254 @@ void main() {
     });
   });
 
+  group('Message row thread navigation', () {
+    testWidgets('a local reply makes its thread immediately tappable', (
+      tester,
+    ) async {
+      final root = _textMsg(
+        id: 'local-root',
+        pubkey: 'alice',
+        content: 'Start a local thread',
+      );
+      final reply = _textMsg(
+        id: 'local-reply',
+        pubkey: 'self',
+        content: 'Waiting for relay echo',
+        createdAt: 1100,
+        extraTags: const [
+          ['e', 'local-root', '', 'reply'],
+        ],
+      );
+      final query = Completer<List<NostrEvent>>();
+      final notifier = _FakeMessagesNotifier([root]);
+      await tester.pumpWidget(
+        _buildTestable(
+          messages: [root],
+          messagesNotifier: notifier,
+          pendingThreadReplies: {'local-root': query.future},
+        ),
+      );
+      await tester.pumpAndSettle();
+      await tester.tap(findRichText('Start a local thread'));
+      await tester.pumpAndSettle();
+      expect(find.byType(ThreadDetailPage), findsNothing);
+
+      notifier.addLocalMessage(reply);
+      notifier.completeLocalMessage('local-reply');
+      await tester.pumpAndSettle();
+      await tester.tap(findRichText('Start a local thread'));
+      await tester.pumpAndSettle();
+      expect(find.byType(ThreadDetailPage), findsOneWidget);
+      expect(findRichText('Waiting for relay echo'), findsOneWidget);
+
+      query.complete([reply]);
+      await tester.pumpAndSettle();
+      await tester.pageBack();
+      await tester.pumpAndSettle();
+      await tester.tap(findRichText('Start a local thread'));
+      await tester.pumpAndSettle();
+      expect(find.byType(ThreadDetailPage), findsOneWidget);
+      expect(findRichText('Waiting for relay echo'), findsOneWidget);
+    });
+
+    testWidgets(
+      'pending recount renders a lower-bound count and remains tappable',
+      (tester) async {
+        final root = _textMsg(
+          id: 'overflow-root',
+          pubkey: 'alice',
+          content: 'Overflow thread',
+        );
+        await tester.pumpWidget(
+          _buildTestable(
+            messages: [root],
+            messagesNotifier: _FakeMessagesNotifier(
+              [root],
+              summaries: const {
+                'overflow-root': ChannelWindowThreadSummary(
+                  replyCount: 256,
+                  descendantCount: 256,
+                  lastReplyAt: 1100,
+                  participantPubkeys: ['bob'],
+                  isLowerBound: true,
+                ),
+              },
+            ),
+            threadReplies: const {'overflow-root': []},
+          ),
+        );
+        await tester.pumpAndSettle();
+        expect(findRichText('256+ replies'), findsOneWidget);
+        await tester.tap(findRichText('Overflow thread'));
+        await tester.pumpAndSettle();
+        expect(find.byType(ThreadDetailPage), findsOneWidget);
+      },
+    );
+
+    testWidgets(
+      'unknown deletion hides the number while preserving thread navigation',
+      (tester) async {
+        final root = _textMsg(
+          id: 'overflow-root',
+          pubkey: 'alice',
+          content: 'Overflow thread',
+        );
+        await tester.pumpWidget(
+          _buildTestable(
+            messages: [root],
+            messagesNotifier: _FakeMessagesNotifier(
+              [root],
+              summaries: const {
+                'overflow-root': ChannelWindowThreadSummary(
+                  replyCount: 256,
+                  descendantCount: 256,
+                  lastReplyAt: 1100,
+                  participantPubkeys: ['bob'],
+                  isCountPending: true,
+                ),
+              },
+            ),
+            threadReplies: const {'overflow-root': []},
+          ),
+        );
+        await tester.pumpAndSettle();
+        expect(findRichText('Replies'), findsOneWidget);
+        await tester.tap(findRichText('Overflow thread'));
+        await tester.pumpAndSettle();
+        expect(find.byType(ThreadDetailPage), findsOneWidget);
+      },
+    );
+
+    testWidgets('tapping a message with unloaded replies opens its thread', (
+      tester,
+    ) async {
+      final root = _textMsg(
+        id: 'remote-root',
+        pubkey: 'alice',
+        content: 'Replies are on the relay',
+      );
+      await tester.pumpWidget(
+        _buildTestable(
+          messages: [root],
+          messagesNotifier: _FakeMessagesNotifier(
+            [root],
+            summaries: const {
+              'remote-root': ChannelWindowThreadSummary(
+                replyCount: 1,
+                descendantCount: 1,
+                lastReplyAt: 1100,
+                participantPubkeys: ['bob'],
+              ),
+            },
+          ),
+          threadReplies: const {'remote-root': []},
+        ),
+      );
+      await tester.pumpAndSettle();
+      await tester.tap(findRichText('Replies are on the relay'));
+      await tester.pumpAndSettle();
+      expect(find.byType(ThreadDetailPage), findsOneWidget);
+    });
+
+    for (final directReplies in [0, 1]) {
+      testWidgets(
+        'cold broadcast row with $directReplies relay-only direct replies',
+        (tester) async {
+          final root = _textMsg(
+            id: 'outer-root',
+            pubkey: 'bob',
+            content: 'Outer root',
+          );
+          final broadcast = _textMsg(
+            id: 'broadcast-reply',
+            pubkey: 'alice',
+            content: 'Broadcast reply with unloaded children',
+            extraTags: const [
+              ['e', 'outer-root', '', 'root'],
+              ['e', 'outer-root', '', 'reply'],
+              ['broadcast', '1'],
+            ],
+          );
+          await tester.pumpWidget(
+            _buildTestable(
+              messages: [root, broadcast],
+              messagesNotifier: _FakeMessagesNotifier(
+                [root, broadcast],
+                summaries: {
+                  'broadcast-reply': ChannelWindowThreadSummary(
+                    replyCount: directReplies,
+                    descendantCount: 0,
+                    lastReplyAt: 1100,
+                    participantPubkeys: const ['bob'],
+                  ),
+                },
+              ),
+              threadReplies: const {'broadcast-reply': []},
+            ),
+          );
+          await tester.pumpAndSettle();
+          await tester.tap(
+            findRichText('Broadcast reply with unloaded children'),
+          );
+          await tester.pumpAndSettle();
+          expect(
+            find.byType(ThreadDetailPage),
+            directReplies > 0 ? findsOneWidget : findsNothing,
+          );
+        },
+      );
+    }
+
+    for (final hasReplies in [false, true]) {
+      testWidgets(
+        hasReplies
+            ? 'tapping a message with replies opens its thread'
+            : 'tapping a message without replies stays in the channel',
+        (tester) async {
+          final root = _textMsg(
+            id: 'tap-root',
+            pubkey: 'alice',
+            content: 'Tap this message',
+          );
+          final replies = [
+            if (hasReplies)
+              _textMsg(
+                id: 'tap-reply',
+                pubkey: 'bob',
+                content: 'A reply',
+                createdAt: 1100,
+                extraTags: const [
+                  ['e', 'tap-root', '', 'reply'],
+                ],
+              ),
+          ];
+          await tester.pumpWidget(
+            _buildTestable(
+              messages: [root, ...replies],
+              threadReplies: {'tap-root': replies},
+            ),
+          );
+          await tester.pumpAndSettle();
+
+          await tester.tap(findRichText('Tap this message'));
+          await tester.pumpAndSettle();
+
+          expect(
+            find.byType(ThreadDetailPage),
+            hasReplies ? findsOneWidget : findsNothing,
+          );
+          if (!hasReplies) {
+            await tester.longPress(findRichText('Tap this message'));
+            await tester.pumpAndSettle();
+            await tester.tap(find.text('Reply'));
+            await tester.pumpAndSettle();
+            expect(find.byType(ThreadDetailPage), findsOneWidget);
+          }
+        },
+      );
+    }
+  });
+
   group('Deep-link navigation', () {
     testWidgets('fades the target highlight in after the thread route lands', (
       tester,
@@ -14347,6 +14607,7 @@ class _FakeMessagesNotifier extends ChannelMessagesNotifier {
   bool _hasLoadedMessages;
   final List<List<NostrEvent>> _olderPages;
   final bool failOlderFetch;
+  final Map<String, ChannelWindowThreadSummary> summaries;
   int fetchOlderCalls = 0;
 
   _FakeMessagesNotifier(
@@ -14355,6 +14616,7 @@ class _FakeMessagesNotifier extends ChannelMessagesNotifier {
     bool hasLoadedMessages = true,
     List<List<NostrEvent>> olderPages = const [],
     this.failOlderFetch = false,
+    this.summaries = const {},
   }) : _hasLoadedMessages = hasLoadedMessages,
        _olderPages = [...olderPages],
        super(channelId);
@@ -14365,6 +14627,9 @@ class _FakeMessagesNotifier extends ChannelMessagesNotifier {
   @override
   bool get hasLoadedMessages => _hasLoadedMessages;
 
+  @override
+  Map<String, ChannelWindowThreadSummary> get threadSummaries => summaries;
+
   @override
   bool get reachedOldest => _olderPages.isEmpty && !failOlderFetch;
 

mobile/test/features/channels/thread_replies_provider_test.dart

@@ -8,6 +8,8 @@ import 'package:hooks_riverpod/hooks_riverpod.dart';
 
 class _FakeRelaySession extends RelaySessionNotifier {
   int queryCount = 0;
+  bool honorDepthLimit = false;
+  final filtersSeen = <NostrFilter>[];
   List<NostrEvent> replies = const [];
   Completer<List<NostrEvent>>? nextQueryGate;
 
@@ -24,11 +26,16 @@ class _FakeRelaySession extends RelaySessionNotifier {
     Duration timeout = const Duration(seconds: 8),
   }) async {
     queryCount++;
+    filtersSeen.addAll(filters);
     final gate = nextQueryGate;
     if (gate != null) {
       nextQueryGate = null;
       return gate.future;
     }
+    if (honorDepthLimit) {
+      final depth = filters.single.extensions['depth_limit'] as int;
+      return replies.where((event) => event.createdAt <= depth).toList();
+    }
     return replies;
   }
 }
@@ -49,6 +56,85 @@ NostrEvent _reply(String id, int createdAt) => NostrEvent(
 void main() {
   const args = ThreadRepliesArgs(channelId: 'chan', rootId: 'root');
 
+  test('complete scan includes a legal 80-deep reply chain', () async {
+    final session = _FakeRelaySession()
+      ..honorDepthLimit = true
+      ..replies = [
+        for (var depth = 1; depth <= 80; depth++)
+          NostrEvent(
+            id: 'reply-$depth',
+            pubkey: 'bob',
+            createdAt: depth,
+            kind: EventKind.streamMessage,
+            tags: [
+              ['h', 'chan'],
+              ['e', 'root', '', 'root'],
+              ['e', depth == 1 ? 'root' : 'reply-${depth - 1}', '', 'reply'],
+            ],
+            content: '',
+            sig: '',
+          ),
+      ];
+    final container = ProviderContainer(
+      overrides: [relaySessionProvider.overrideWith(() => session)],
+    );
+    addTearDown(container.dispose);
+    container.listen(threadRepliesProvider(args), (_, _) {});
+    expect(
+      await container.read(threadRepliesProvider(args).future),
+      hasLength(80),
+    );
+  });
+
+  test(
+    'origin cursor includes epoch-zero events in an exhaustive scan',
+    () async {
+      final session = _FakeRelaySession()..replies = [_reply('epoch', 0)];
+      final container = ProviderContainer(
+        overrides: [relaySessionProvider.overrideWith(() => session)],
+      );
+      addTearDown(container.dispose);
+      container.listen(threadRepliesProvider(args), (_, _) {});
+      expect(
+        (await container.read(
+          threadRepliesProvider(args).future,
+        )).single.createdAt,
+        0,
+      );
+      expect(session.filtersSeen.single.extensions['thread_cursor'], -1);
+      expect(
+        session.filtersSeen.single.extensions['thread_cursor_id'],
+        '0' * 64,
+      );
+    },
+  );
+
+  test(
+    'disposes empty local overlays after their last listener leaves',
+    () async {
+      final container = ProviderContainer();
+      addTearDown(container.dispose);
+      final provider = threadLocalRepliesProvider(args);
+      final subscription = container.listen(provider, (_, _) {});
+      subscription.close();
+      await container.pump();
+      expect(container.exists(provider), isFalse);
+    },
+  );
+
+  test('retains local replies without listeners until confirmation', () async {
+    final container = ProviderContainer();
+    addTearDown(container.dispose);
+    final provider = threadLocalRepliesProvider(args);
+    final notifier = container.read(provider.notifier);
+    notifier.add(_reply('pending', 1000));
+    await container.pump();
+    expect(container.read(provider).single.id, 'pending');
+    notifier.confirm({'pending'});
+    await container.pump();
+    expect(container.exists(provider), isFalse);
+  });
+
   (ProviderContainer, _FakeRelaySession, ProviderSubscription<Object?>)
   makeHarness(List<NostrEvent> initialReplies) {
     final fakeSession = _FakeRelaySession()..replies = initialReplies;

mobile/test/features/channels/thread_reply_ownership_test.dart

@@ -0,0 +1,30 @@
+import 'package:buzz/features/channels/thread_reply_ownership.dart';
+import 'package:buzz/shared/relay/relay.dart';
+import 'package:flutter_test/flutter_test.dart';
+
+void main() {
+  test(
+    'identity evidence is bounded while keeping recent evicted-payload owners',
+    () {
+      final owners = ThreadReplyOwnership();
+      owners.record([
+        for (var i = 0; i < 8193; i++)
+          NostrEvent(
+            id: 'reply-$i',
+            pubkey: 'author',
+            createdAt: i,
+            kind: EventKind.streamMessage,
+            tags: const [
+              ['h', 'channel'],
+              ['e', 'root', '', 'reply'],
+            ],
+            content: '',
+            sig: '',
+          ),
+      ]);
+      expect(owners.rootFor('reply-0'), isNull);
+      expect(owners.rootFor('reply-1'), 'root');
+      expect(owners.rootFor('reply-8192'), 'root');
+    },
+  );
+}

mobile/test/features/channels/timeline_message_test.dart

@@ -1050,6 +1050,28 @@ void main() {
       expect(entries.single.summary!.replyCount, 5);
     });
 
+    test('relay direct replies are a lower bound for a broadcast row', () {
+      final messages = formatTimeline([
+        _textMsg(id: 'a'),
+        _replyMsg(
+          id: 'broadcast',
+          parentId: 'a',
+          extraTags: const [
+            ['broadcast', '1'],
+          ],
+        ),
+      ]);
+      final entries = buildMainTimelineEntries(
+        messages,
+        relaySummaries: {
+          'broadcast': relaySummary(replyCount: 1, descendantCount: 0),
+        },
+      );
+      final summary = entries.last.summary!;
+      expect(summary.replyCount, 1);
+      expect(summary.isLowerBound, isTrue);
+    });
+
     test('a nested reply badges the reply it answers', () {
       final messages = formatTimeline([
         _textMsg(id: 'a', createdAt: 1000),