diff --git a/apps/buzz-gpui/crates/app/src/agent_activity/mod.rs b/apps/buzz-gpui/crates/app/src/agent_activity/mod.rs index 764a1a50cb2..07a473da441 100644 --- a/apps/buzz-gpui/crates/app/src/agent_activity/mod.rs +++ b/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; diff --git a/apps/buzz-gpui/crates/app/src/agent_activity/request.rs b/apps/buzz-gpui/crates/app/src/agent_activity/request.rs index 564936fdff4..a5a43c502cd 100644 --- a/apps/buzz-gpui/crates/app/src/agent_activity/request.rs +++ b/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>>; + +/// A request-work question: the channel and the message ids. +type WorkKey = (String, Vec); + +/// 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>>, + messages: Mutex>, +} + +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( + memo: &Mutex>, + 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 { + 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 { let live: Vec> = 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 diff --git a/apps/buzz-gpui/crates/app/src/agent_activity/request/tests.rs b/apps/buzz-gpui/crates/app/src/agent_activity/request/tests.rs index ff7a860151f..6b8c751d07d 100644 --- a/apps/buzz-gpui/crates/app/src/agent_activity/request/tests.rs +++ b/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!(©, s); +} diff --git a/apps/buzz-gpui/crates/app/src/ui/agent_activity/bubble.rs b/apps/buzz-gpui/crates/app/src/ui/agent_activity/bubble.rs index 9db2978931b..f91cedba826 100644 --- a/apps/buzz-gpui/crates/app/src/ui/agent_activity/bubble.rs +++ b/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, pub channel: String, - pub works: Arc>>, + pub works: request::MessageWorks, } impl Bubbles { /// Every message in `channel` with work to show. pub fn of(community: &Entity, 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), } } }