perf(gpui): compute agent activity bubbles only when activity changes

apps/buzz-gpui/crates/app/src/agent_activity/mod.rs

@@ -130,7 +130,8 @@ pub enum Capture {
     Error,
 }
 
-/// What views read; replaced as a whole when anything changes.
+/// What views read; replaced as a whole when anything changes, never
+/// edited in place, so work worked out from one stays true for its life.
 #[derive(Clone, Debug, Default, PartialEq)]
 pub struct Snapshot {
     pub status: FeedStatus,
@@ -155,6 +156,9 @@ pub struct Snapshot {
     /// `(channel, message)` pairs live turn starts name as their triggers:
     /// the messages telemetry may tie work to.
     pub requested: Arc<[(String, String)]>,
+    /// Request work already worked out from this snapshot: views ask on
+    /// every frame, the answer changes only with a new snapshot.
+    pub(crate) memo: request::Memo,
 }
 
 /// One page read from the archive, decrypted.
@@ -696,6 +700,7 @@ impl Activity {
             capture: self.capture,
             working_channels: working.into(),
             requested,
+            memo: request::Memo::default(),
         };
         if next == self.snapshot {
             return false;

apps/buzz-gpui/crates/app/src/agent_activity/request.rs

@@ -8,9 +8,69 @@ use super::local::{LocalWork, LocalWorkState, WorkStep, breathing, local_activit
 use super::transcript_model::{ItemKind, TranscriptTurn};
 use super::turns::{ActivityTurn, TurnState, parse_time};
 use super::{FeedStatus, Record, Snapshot, raw_records};
+use buzz_relay_client::EventId;
 use serde_json::Value;
 use std::collections::{BTreeSet, HashMap, HashSet};
-use std::sync::Arc;
+use std::hash::Hash;
+use std::sync::{Arc, Mutex};
+
+/// Each message in a channel with work to show, as its bubble shows it.
+pub type MessageWorks = Arc<HashMap<EventId, Vec<RequestWork>>>;
+
+/// A request-work question: the channel and the message ids.
+type WorkKey = (String, Vec<String>);
+
+/// Answers the cap allows one snapshot to remember; past it the memo
+/// starts over.
+const MEMO_LIMIT: usize = 256;
+
+/// Request work worked out from one [`Snapshot`]. Reading records is
+/// costly and views ask on every frame while working dots animate, but a
+/// snapshot never changes, so its answers hold until the next one. A
+/// clone starts empty, and memos never make snapshots differ.
+#[derive(Default)]
+pub struct Memo {
+    work: Mutex<HashMap<WorkKey, Vec<RequestWork>>>,
+    messages: Mutex<HashMap<String, MessageWorks>>,
+}
+
+impl Clone for Memo {
+    fn clone(&self) -> Self {
+        Self::default()
+    }
+}
+
+impl PartialEq for Memo {
+    fn eq(&self, _: &Self) -> bool {
+        true
+    }
+}
+
+impl std::fmt::Debug for Memo {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("Memo")
+    }
+}
+
+/// The remembered answer for `key`, else `work`'s, then remembered. The
+/// lock is not held while working.
+fn remembered<K: Eq + Hash, V: Clone>(
+    memo: &Mutex<HashMap<K, V>>,
+    key: K,
+    work: impl FnOnce() -> V,
+) -> V {
+    let lock = || memo.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+    if let Some(value) = lock().get(&key) {
+        return value.clone();
+    }
+    let value = work();
+    let mut memo = lock();
+    if memo.len() >= MEMO_LIMIT {
+        memo.clear();
+    }
+    memo.insert(key, value.clone());
+    value
+}
 
 /// A request's state, from local work or from telemetry.
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
@@ -216,8 +276,19 @@ pub struct RequestWork {
     pub working: bool,
 }
 
-/// Each agent handling any of `ids` in `channel`, sorted by key.
+/// Each agent handling any of `ids` in `channel`, sorted by key;
+/// remembered for the snapshot.
 pub fn request_work(snapshot: &Snapshot, channel: &str, ids: &[&str]) -> Vec<RequestWork> {
+    let key: WorkKey = (
+        channel.to_string(),
+        ids.iter().map(|id| id.to_string()).collect(),
+    );
+    remembered(&snapshot.memo.work, key, || {
+        work_out(snapshot, channel, ids)
+    })
+}
+
+fn work_out(snapshot: &Snapshot, channel: &str, ids: &[&str]) -> Vec<RequestWork> {
     let live: Vec<Arc<Record>> = snapshot
         .records
         .iter()
@@ -356,6 +427,23 @@ pub fn message_activity(snapshot: &Snapshot, channel: &str, message: &str) -> Ve
         .collect()
 }
 
+/// Every message in `channel` with work to show, by message, remembered
+/// for the snapshot.
+pub fn message_works(snapshot: &Snapshot, channel: &str) -> MessageWorks {
+    remembered(&snapshot.memo.messages, channel.to_string(), || {
+        Arc::new(
+            candidates(snapshot, channel)
+                .into_iter()
+                .filter_map(|id| {
+                    let works = message_activity(snapshot, channel, &id);
+                    (!works.is_empty()).then(|| Some((EventId::from_hex(&id).ok()?, works)))
+                })
+                .flatten()
+                .collect(),
+        )
+    })
+}
+
 /// Agents working in the thread `root` (`threadWorkingAgents`), from the
 /// same request state as the bubbles. `represented` is a message whose
 /// own bubble already shows these agents, so the thread summary leaves

apps/buzz-gpui/crates/app/src/agent_activity/request/tests.rs

@@ -238,3 +238,65 @@ fn a_recovered_session_shows_working_over_the_failure_it_replaced() {
     assert!(f.bubble(&mention)[0].working);
     assert_eq!(f.thread(&[&agent]), vec![agent.clone()]);
 }
+
+#[test]
+fn a_snapshot_remembers_its_request_work_until_replaced() {
+    let (request, root) = (key('c'), key('b'));
+    let mut f = Fixture::new();
+    f.send("turn_started", "one", &request);
+    f.message(&request, &root);
+    f.send("turn_liveness", "one", &request);
+    let s = f.snapshot();
+    assert_eq!(
+        request_work(s, "channel", &[&request]),
+        work_out(s, "channel", &[&request])
+    );
+    // A remembered answer is served without reading records again.
+    s.memo
+        .work
+        .lock()
+        .unwrap()
+        .insert(("channel".into(), vec![request.clone()]), Vec::new());
+    assert!(f.bubble(&request).is_empty());
+    assert!(bubbled_agents(f.snapshot(), "channel", None).is_empty());
+    // New activity replaces the snapshot, and with it the memo.
+    f.send("turn_liveness", "one", &request);
+    assert!(f.bubble(&request)[0].working);
+    assert_eq!(
+        bubbled_agents(f.snapshot(), "channel", None),
+        vec![key('a')]
+    );
+}
+
+#[test]
+fn message_works_are_shared_until_activity_changes() {
+    let request = key('c');
+    let mut f = Fixture::new();
+    f.send("turn_started", "one", &request);
+    f.send("turn_liveness", "one", &request);
+    let first = message_works(f.snapshot(), "channel");
+    let id = EventId::from_hex(&request).unwrap();
+    assert!(first[&id][0].working);
+    assert_eq!(first.len(), 1);
+    assert!(Arc::ptr_eq(&first, &message_works(f.snapshot(), "channel")));
+    assert!(message_works(f.snapshot(), "elsewhere").is_empty());
+    f.send("turn_completed", "one", &request);
+    let after = message_works(f.snapshot(), "channel");
+    assert!(!Arc::ptr_eq(&first, &after));
+    assert!(after.is_empty());
+}
+
+#[test]
+fn a_memo_is_bounded_and_never_copied_or_compared() {
+    let request = key('c');
+    let mut f = Fixture::new();
+    f.send("turn_started", "one", &request);
+    let s = f.snapshot();
+    for n in 0..MEMO_LIMIT + 10 {
+        request_work(s, &format!("channel-{n}"), &[&request]);
+    }
+    assert!(s.memo.work.lock().unwrap().len() <= MEMO_LIMIT);
+    let copy = s.clone();
+    assert!(copy.memo.work.lock().unwrap().is_empty());
+    assert_eq!(&copy, s);
+}

apps/buzz-gpui/crates/app/src/ui/agent_activity/bubble.rs

@@ -17,33 +17,23 @@ use buzz_ui::{
 };
 use gpui_kit::prelude::FluentBuilder;
 use gpui_kit::*;
-use std::collections::HashMap;
-use std::sync::Arc;
 
 /// The bubbles of one conversation's rows, by message.
 #[derive(Clone)]
 pub struct Bubbles {
     pub community: WeakEntity<Community>,
     pub channel: String,
-    pub works: Arc<HashMap<EventId, Vec<RequestWork>>>,
+    pub works: request::MessageWorks,
 }
 
 impl Bubbles {
     /// Every message in `channel` with work to show.
     pub fn of(community: &Entity<Community>, channel: &str, cx: &App) -> Self {
         let snapshot = observer::snapshot(community, cx);
-        let works = request::candidates(&snapshot, channel)
-            .into_iter()
-            .filter_map(|id| {
-                let works = message_activity(&snapshot, channel, &id);
-                (!works.is_empty()).then(|| Some((EventId::from_hex(&id).ok()?, works)))
-            })
-            .flatten()
-            .collect();
         Self {
             community: community.downgrade(),
             channel: channel.to_string(),
-            works: Arc::new(works),
+            works: request::message_works(&snapshot, channel),
         }
     }
 }