From c665b0bbbaad926aecc843e970ccb64c3e48092e Mon Sep 17 00:00:00 2001 From: hoangvu12 Date: Sat, 17 Oct 2026 07:31:09 +0700 Subject: [PATCH] fix(engine): bound idle git work by checkout, not by subscriber PR-status watches re-ran the ~8-process checkout inspection once per subscription every 15 s, so cost grew with chats x clients (a paired phone watches up to 40 targets regardless of the desktop's hidden-badge setting). Workspace-file authorization ran the full refs() (~5 processes) on every request. - Share one single-flight inspection per checkout and skip it while a stamp of HEAD/config/packed-refs is unchanged (5 min max age). The locked section runs detached so an unpolled subscriber cannot strand the lock. - Authorize checkouts from one cached `git worktree list`, revalidated against the mtime of .git/worktrees (10 s TTL otherwise). - Desktop: watch side chats only while they or their parent are selected; reconcile on selection. --- crates/engine/src/change_requests.rs | 239 ++++++++++++++++++++++++++- crates/engine/src/repos.rs | 147 +++++++++++++++- crates/ui/src/change_requests.rs | 30 +++- crates/ui/src/state.rs | 83 +++++++++- 4 files changed, 478 insertions(+), 21 deletions(-) diff --git a/crates/engine/src/change_requests.rs b/crates/engine/src/change_requests.rs index 33f17d1420..e06d40c48b 100644 --- a/crates/engine/src/change_requests.rs +++ b/crates/engine/src/change_requests.rs @@ -25,6 +25,12 @@ const WITHOUT_CHANGE_REQUEST_TTL: Duration = Duration::from_secs(45); const FAILURE_INITIAL_BACKOFF: Duration = Duration::from_secs(20); const FAILURE_MAX_BACKOFF: Duration = Duration::from_secs(15 * 60); const CONTEXT_POLL_INTERVAL: Duration = Duration::from_secs(15); +/// An inspection whose Git metadata stamp is unchanged is reused up to this +/// age. The stamp misses inputs outside the checkout (global config, +/// loose remote-tracking refs), so it still expires. +const CONTEXT_MAX_AGE: Duration = Duration::from_secs(5 * 60); +/// A stamp on a wedged mount degrades to the poll interval instead of hanging. +const GIT_STAMP_TIMEOUT: Duration = Duration::from_secs(2); #[derive(Debug, Clone, PartialEq, Eq, Hash)] pub struct ChangeRequestCacheKey { @@ -41,6 +47,7 @@ struct Timing { failure_initial_backoff: Duration, failure_max_backoff: Duration, context_poll_interval: Duration, + context_max_age: Duration, } impl Default for Timing { @@ -51,6 +58,7 @@ impl Default for Timing { failure_initial_backoff: FAILURE_INITIAL_BACKOFF, failure_max_backoff: FAILURE_MAX_BACKOFF, context_poll_interval: CONTEXT_POLL_INTERVAL, + context_max_age: CONTEXT_MAX_AGE, } } } @@ -67,11 +75,60 @@ struct CacheState { last_error: Option, } +/// One checkout's latest Git inspection, shared by every subscription on +/// that checkout whatever branch it asks about. The lock makes concurrent +/// pollers wait for one inspection instead of each spawning their own. +struct CheckoutInspection { + latest: AsyncMutex>, +} + +struct InspectedCheckout { + source: Result, + started: Instant, + stamp: Option, +} + +/// Size and mtime of the files an inspection reads: the worktree's `HEAD` +/// (branch switches) and the shared `config` and `packed-refs` (upstream, +/// remote and push-default changes). +#[derive(Debug, Clone, PartialEq, Eq)] +struct GitStamp(Vec>); + +impl GitStamp { + async fn read(git_dir: &Path) -> Option { + let git_dir = git_dir.to_owned(); + let read = tokio::task::spawn_blocking(move || Self::read_blocking(&git_dir)); + tokio::time::timeout(GIT_STAMP_TIMEOUT, read) + .await + .ok()? + .ok()? + } + + fn read_blocking(git_dir: &Path) -> Option { + let stat = |path: PathBuf| { + let metadata = std::fs::metadata(path).ok()?; + Some((metadata.len(), metadata.modified().ok()?)) + }; + // A linked worktree's git dir names the shared one in `commondir`. + let common_dir = std::fs::read_to_string(git_dir.join("commondir")) + .map(|relative| git_dir.join(relative.trim())) + .unwrap_or_else(|_| git_dir.to_owned()); + let head = stat(git_dir.join("HEAD"))?; + Some(Self(vec![ + Some(head), + stat(common_dir.join("config")), + stat(common_dir.join("packed-refs")), + ])) + } +} + struct Inner { repos: Repos, device_id: String, lookup: Arc, entries: Mutex>>, + /// Keyed by checkout id; the subscriptions hold the strong references. + inspections: Mutex>>, timing: Timing, shutdown: CancellationToken, } @@ -108,6 +165,7 @@ impl CheckoutChangeRequests { device_id: device_id.to_owned(), lookup, entries: Mutex::new(HashMap::new()), + inspections: Mutex::new(HashMap::new()), timing, shutdown: CancellationToken::new(), }), @@ -155,6 +213,7 @@ impl CheckoutChangeRequests { ) -> BoxStream<'static, CheckoutChangeRequestStatus> { let state = SubscriptionState { service: self.clone(), + inspection: self.inspection_for(&identity.id), cwd, identity, requested_branch, @@ -168,9 +227,7 @@ impl CheckoutChangeRequests { } let mut source = match state .service - .inner - .lookup - .inspect_checkout(&state.cwd) + .inspect(&state.inspection, &state.cwd, &state.identity.git_dir) .await { Ok(source) => source, @@ -218,6 +275,77 @@ impl CheckoutChangeRequests { .boxed() } + fn inspection_for(&self, checkout_id: &str) -> Arc { + let mut inspections = self + .inner + .inspections + .lock() + .unwrap_or_else(PoisonError::into_inner); + if let Some(inspection) = inspections.get(checkout_id).and_then(Weak::upgrade) { + return inspection; + } + inspections.retain(|_, inspection| inspection.strong_count() > 0); + let inspection = Arc::new(CheckoutInspection { + latest: AsyncMutex::new(None), + }); + inspections.insert(checkout_id.to_owned(), Arc::downgrade(&inspection)); + inspection + } + + /// The checkout's Git context, inspected at most once per poll interval + /// for all of its subscribers, and not at all while the stamp of its Git + /// metadata is unchanged (up to `context_max_age`). + async fn inspect( + &self, + inspection: &Arc, + cwd: &Path, + git_dir: &Path, + ) -> Result { + // Detached: a subscriber that stops polling mid-inspection keeps its + // pending future, and with it the lock every other subscriber on the + // checkout waits for. The task always finishes and releases it. + let service = self.clone(); + let inspection = inspection.clone(); + let (cwd, git_dir) = (cwd.to_owned(), git_dir.to_owned()); + tokio::spawn(async move { service.inspect_shared(&inspection, &cwd, &git_dir).await }) + .await + .unwrap_or(Err(ChangeRequestError::RepositoryUnavailable)) + } + + async fn inspect_shared( + &self, + inspection: &CheckoutInspection, + cwd: &Path, + git_dir: &Path, + ) -> Result { + let timing = self.inner.timing; + let mut latest = inspection.latest.lock().await; + if let Some(previous) = latest.as_ref() + && previous.started.elapsed() < timing.context_poll_interval + { + return previous.source.clone(); + } + let stamp = GitStamp::read(git_dir).await; + if let Some(previous) = latest.as_ref() + && previous.source.is_ok() + && stamp.is_some() + && stamp == previous.stamp + && previous.started.elapsed() < timing.context_max_age + { + return previous.source.clone(); + } + // Stamped before inspecting: a change racing the inspection leaves + // the stamps different, so the next poll inspects again. + let started = Instant::now(); + let source = self.inner.lookup.inspect_checkout(cwd).await; + *latest = Some(InspectedCheckout { + source: source.clone(), + started, + stamp, + }); + source + } + fn acquire(&self, key: ChangeRequestCacheKey) -> DemandLease { let entry = { let mut entries = self @@ -340,6 +468,7 @@ impl Drop for DemandLease { struct SubscriptionState { service: CheckoutChangeRequests, + inspection: Arc, cwd: PathBuf, identity: CheckoutIdentity, requested_branch: Option, @@ -421,6 +550,7 @@ mod tests { source: Mutex, results: Mutex, ChangeRequestError>>>, resolves: AtomicUsize, + inspections: AtomicUsize, } impl FakeLookup { @@ -432,9 +562,16 @@ mod tests { source: Mutex::new(source), results: Mutex::new(results.into_iter().collect()), resolves: AtomicUsize::new(0), + inspections: AtomicUsize::new(0), }) } + /// Each inspection stands for the ~8 `git` processes the real + /// inspector spawns. + fn inspection_count(&self) -> usize { + self.inspections.load(Ordering::Acquire) + } + fn set_source(&self, source: CheckoutSourceContext) { *self.source.lock().unwrap() = source; } @@ -450,6 +587,7 @@ mod tests { &self, _cwd: &Path, ) -> Result { + self.inspections.fetch_add(1, Ordering::AcqRel); Ok(self.source.lock().unwrap().clone()) } @@ -516,6 +654,7 @@ mod tests { failure_initial_backoff: Duration::from_secs(1), failure_max_backoff: Duration::from_secs(2), context_poll_interval: Duration::from_millis(5), + context_max_age: Duration::from_millis(5), } } @@ -655,6 +794,7 @@ mod tests { failure_initial_backoff: Duration::from_millis(1), failure_max_backoff: Duration::from_millis(1), context_poll_interval: Duration::from_millis(1), + context_max_age: Duration::from_millis(1), }; let service = service(lookup.clone(), timing); let key = cache_key("checkout-1", &source); @@ -732,4 +872,97 @@ mod tests { assert!(stream.next().await.is_none()); } + + fn no_pull_requests() + -> impl Iterator, ChangeRequestError>> { + std::iter::repeat_with(|| Ok(None)).take(1_000) + } + + /// Drive `streams` in the background so they poll like live RPC + /// subscriptions, returning the tasks to join after shutdown. + fn drain( + streams: impl IntoIterator>, + ) -> Vec> { + streams + .into_iter() + .map(|mut stream| tokio::spawn(async move { while stream.next().await.is_some() {} })) + .collect() + } + + /// Several chats on one checkout (an orchestrator's side chats, or a + /// desktop and a phone watching the same branch) must not each run the + /// checkout's Git inspection: one inspection per poll interval serves + /// every subscriber. + #[tokio::test(start_paused = true)] + async fn watchers_on_one_checkout_share_one_inspection_per_interval() { + const WATCHERS: usize = 8; + const INTERVALS: u32 = 10; + let lookup = FakeLookup::new(source("feature/status"), no_pull_requests()); + let service = service(lookup.clone(), Timing::default()); + let tasks = drain((0..WATCHERS).map(|index| { + service.watch_checkout_for_branch( + PathBuf::from("/checkout"), + identity(), + Some(format!("feature/{index}")), + ) + })); + + tokio::time::sleep(CONTEXT_POLL_INTERVAL * INTERVALS + Duration::from_secs(1)).await; + let inspections = lookup.inspection_count(); + service.shutdown(); + for task in tasks { + task.await.unwrap(); + } + + assert!( + inspections <= INTERVALS as usize + 1, + "{WATCHERS} watchers ran {inspections} checkout inspections over {INTERVALS} poll intervals" + ); + } + + /// While HEAD, the config and packed refs are untouched, an idle + /// checkout is not re-inspected at all; moving HEAD is picked up on the + /// next poll. + #[tokio::test(start_paused = true)] + async fn unchanged_git_metadata_skips_reinspection_until_head_moves() { + let git_dir = tempfile::tempdir().unwrap(); + std::fs::write(git_dir.path().join("HEAD"), "ref: refs/heads/feature/one\n").unwrap(); + std::fs::write(git_dir.path().join("config"), "[core]\n").unwrap(); + let identity = CheckoutIdentity { + git_dir: git_dir.path().to_owned(), + ..identity() + }; + let lookup = FakeLookup::new(source("feature/one"), no_pull_requests()); + let service = service(lookup.clone(), Timing::default()); + let mut live = service.watch_checkout(PathBuf::from("/checkout"), identity.clone()); + assert_eq!(live.next().await.unwrap().branch, "feature/one"); + let tasks = drain( + (0..3).map(|_| service.watch_checkout(PathBuf::from("/checkout"), identity.clone())), + ); + + tokio::time::sleep(CONTEXT_POLL_INTERVAL * 10).await; + assert_eq!( + lookup.inspection_count(), + 1, + "an idle checkout was re-inspected" + ); + + std::fs::write( + git_dir.path().join("HEAD"), + "ref: refs/heads/feature/two-and-longer\n", + ) + .unwrap(); + lookup.set_source(source("feature/two-and-longer")); + let switched = tokio::time::timeout(CONTEXT_POLL_INTERVAL * 2, live.next()) + .await + .expect("a HEAD move is seen within one poll interval") + .unwrap(); + assert_eq!(switched.branch, "feature/two-and-longer"); + assert_eq!(lookup.inspection_count(), 2); + + service.shutdown(); + for task in tasks { + task.await.unwrap(); + } + } } diff --git a/crates/engine/src/repos.rs b/crates/engine/src/repos.rs index cb0e2a4411..e8b7969884 100644 --- a/crates/engine/src/repos.rs +++ b/crates/engine/src/repos.rs @@ -43,6 +43,10 @@ const FILE_SEARCH_MAX_RESULTS: usize = 8; const FILE_SEARCH_TIMEOUT: Duration = Duration::from_secs(6); const GITHUB_AVATAR_TIMEOUT: Duration = Duration::from_secs(6); const FILE_INDEX_TTL: Duration = Duration::from_secs(10); +/// Longest a `git worktree list` answers checkout authorization. A change to +/// `.git/worktrees` (`worktree add`/`remove`/`prune`) invalidates it sooner; +/// the TTL covers `worktree move` and repositories it cannot stamp. +const WORKTREE_LIST_TTL: Duration = Duration::from_secs(10); const FILE_INDEX_MAX_ENTRIES: usize = 250_000; const RANK_BUFFER: usize = 1_024; pub const GIT_HISTORY_DEFAULT_LIMIT: usize = 100; @@ -138,6 +142,10 @@ struct ReposInner { github_avatars: std::sync::Mutex>, github_avatar_pages: std::sync::Mutex>, file_index: FileIndexCache, + worktree_lists: std::sync::Mutex>, + /// Every `git` subprocess this instance spawned; tests assert on it. + #[cfg(test)] + git_runs: std::sync::atomic::AtomicUsize, } struct IndexedPath { @@ -153,6 +161,14 @@ struct FileIndex { type FileIndexCache = std::sync::Mutex>>; +/// A repository's linked worktrees as of `listed`, with the mtime of its +/// `.git/worktrees` directory at that moment. +struct WorktreeList { + paths: Vec, + stamp: Option, + listed: std::time::Instant, +} + #[derive(Clone)] pub struct Repos { inner: std::sync::Arc, @@ -185,10 +201,18 @@ impl Repos { github_avatars: std::sync::Mutex::new(HashMap::new()), github_avatar_pages: std::sync::Mutex::new(HashSet::new()), file_index: std::sync::Mutex::new(HashMap::new()), + worktree_lists: std::sync::Mutex::new(HashMap::new()), + #[cfg(test)] + git_runs: std::sync::atomic::AtomicUsize::new(0), }), } } + #[cfg(test)] + fn git_runs(&self) -> usize { + self.inner.git_runs.load(Ordering::Relaxed) + } + // ── registry (repos.json) ─────────────────────────────────────────────── fn registry_path(&self) -> PathBuf { @@ -222,6 +246,8 @@ impl Repos { /// Run `git ` (optionally under `cwd`), returning trimmed stdout. async fn git(&self, args: &[&str], cwd: Option<&Path>) -> Result { + #[cfg(test)] + self.inner.git_runs.fetch_add(1, Ordering::Relaxed); let mut cmd = tokio::process::Command::new("git"); #[cfg(windows)] { @@ -1062,15 +1088,9 @@ impl Repos { /// worktrees. Filesystem resolution happens on a disposable thread because /// user-selected paths may be dead mounts. pub async fn workspace_checkout(&self, repo_path: &Path, candidate: &Path) -> Option { + let worktrees = self.linked_worktrees(repo_path).await; let repo_path = repo_path.to_path_buf(); let candidate = candidate.to_path_buf(); - let worktrees: Vec<_> = self - .refs(&repo_path) - .await - .unwrap_or_default() - .into_iter() - .filter_map(|row| row.worktree_path.map(PathBuf::from)) - .collect(); disposable_worker("checkout-auth", move || { let candidate = std::fs::canonicalize(candidate).ok()?; std::iter::once(repo_path) @@ -1083,6 +1103,43 @@ impl Repos { .flatten() } + /// Linked worktree paths of `repo_path`. Workspace-file requests + /// authorize their checkout on every call, so one `git worktree list` + /// serves them until `.git/worktrees` changes or [`WORKTREE_LIST_TTL`] + /// passes. + async fn linked_worktrees(&self, repo_path: &Path) -> Vec { + let worktrees_dir = repo_path.join(".git").join("worktrees"); + let stamp = disposable_worker("checkout-auth", move || { + std::fs::metadata(worktrees_dir).and_then(|metadata| metadata.modified()) + }) + .await + .and_then(Result::ok); + if let Ok(lists) = self.inner.worktree_lists.lock() + && let Some(list) = lists.get(repo_path) + && list.stamp == stamp + && list.listed.elapsed() < WORKTREE_LIST_TTL + { + return list.paths.clone(); + } + let listed = std::time::Instant::now(); + let paths = self + .git(&["worktree", "list", "--porcelain"], Some(repo_path)) + .await + .map(|out| linked_worktree_paths(&out)) + .unwrap_or_default(); + if let Ok(mut lists) = self.inner.worktree_lists.lock() { + lists.insert( + repo_path.to_path_buf(), + WorktreeList { + paths: paths.clone(), + stamp, + listed, + }, + ); + } + paths + } + /// Switch the checkout at `cwd` (a main folder OR a linked worktree) to /// `ref_name` — the t3code `switchRef` port: an existing local branch is /// checked out directly; a remote-only branch gets a local tracking @@ -1462,6 +1519,17 @@ async fn disposable_worker( rx.await.ok() } +/// Linked worktrees in `git worktree list --porcelain` output: every +/// `worktree ` stanza after the first, which is the main checkout. +fn linked_worktree_paths(porcelain: &str) -> Vec { + porcelain + .lines() + .filter_map(|line| line.trim().strip_prefix("worktree ")) + .skip(1) + .map(PathBuf::from) + .collect() +} + /// The blocking walk: ONE readdir of the target; `is_repo` is a cheap `.git` /// existence probe per directory entry. fn list_folders_blocking(target: &Path, show_hidden: bool) -> Result { @@ -2598,6 +2666,71 @@ tmpfs /run tmpfs rw 0 0 ); } + fn git_fixture(cwd: &Path, args: &[&str]) { + let status = std::process::Command::new("git") + .args([ + "-c", + "user.name=Roboco", + "-c", + "user.email=roboco@example.com", + ]) + .args(args) + .current_dir(cwd) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .status() + .unwrap(); + assert!(status.success(), "git {args:?}"); + } + + /// Workspace-file requests authorize their checkout on every call. That + /// must not re-list branches and worktrees each time: one `git worktree + /// list` serves every request until `.git/worktrees` changes. + #[tokio::test] + async fn workspace_checkout_lists_worktrees_once_until_they_change() { + const REQUESTS: usize = 10; + let data = tempfile::tempdir().unwrap(); + let root = tempfile::tempdir().unwrap(); + let repo = root.path().join("repo"); + std::fs::create_dir(&repo).unwrap(); + git_fixture(&repo, &["init", "-b", "main"]); + git_fixture(&repo, &["commit", "--allow-empty", "-m", "init"]); + let first = root.path().join("first"); + git_fixture(&repo, &["worktree", "add", "-b", "first", "../first"]); + let repos = + Repos::with_worktrees_root(data.path(), "device", data.path().join("worktrees")); + + for _ in 0..REQUESTS { + assert_eq!( + repos.workspace_checkout(&repo, &first).await, + std::fs::canonicalize(&first).ok() + ); + assert!(repos.workspace_checkout(&repo, root.path()).await.is_none()); + } + assert_eq!( + repos.git_runs(), + 1, + "{} authorizations against one repository spawned {} git processes", + REQUESTS * 2, + repos.git_runs() + ); + + // A worktree added after the listing is authorized immediately. + let second = root.path().join("second"); + git_fixture(&repo, &["worktree", "add", "-b", "second", "../second"]); + assert_eq!( + repos.workspace_checkout(&repo, &second).await, + std::fs::canonicalize(&second).ok() + ); + assert_eq!(repos.git_runs(), 2); + + // So is a removed one revoked. + git_fixture(&repo, &["worktree", "remove", "../first"]); + std::fs::create_dir(&first).unwrap(); + assert!(repos.workspace_checkout(&repo, &first).await.is_none()); + assert_eq!(repos.git_runs(), 3); + } + #[tokio::test] async fn concurrent_searches_of_one_checkout_do_not_cancel_each_other() { let data = tempfile::tempdir().unwrap(); diff --git a/crates/ui/src/change_requests.rs b/crates/ui/src/change_requests.rs index 7b30d3e3b8..a73ba0ce7e 100644 --- a/crates/ui/src/change_requests.rs +++ b/crates/ui/src/change_requests.rs @@ -287,14 +287,27 @@ impl ChangeRequestClientState { } /// Active, fully identified checkouts that need host-side PR resolution. +/// +/// Every target is a host-side Git polling loop, so only chats that can show +/// a badge get one: top-level rows, and side chats only while they or their +/// parent are selected (the composer footer and the child-session rows). +/// Orchestrators spawn side chats by the dozen; watching them all made the +/// host's Git work grow with every delegation. pub(crate) fn desired_watch_targets( chats: &[Chat], _spaces: &[Space], + selected_chat: Option<&str>, is_unsupported: impl Fn(&str) -> bool, ) -> HashSet { chats .iter() .filter(|chat| !chat.archived) + .filter(|chat| match chat.parent_chat_id.as_deref() { + None => true, + Some(parent) => { + selected_chat == Some(chat.id.as_str()) || selected_chat == Some(parent) + } + }) .filter(|chat| !is_unsupported(&chat.device_id)) .filter_map(|chat| { let source = chat.source_context.as_ref()?; @@ -458,7 +471,7 @@ mod tests { fn shared_checkout_keeps_conversation_branches_independent() { let first = with_source(chat("first", "local", Some("/repo"), None), "feature/one"); let second = with_source(chat("second", "local", Some("/repo"), None), "feature/two"); - let targets = desired_watch_targets(&[first.clone(), second.clone()], &[], |_| false); + let targets = desired_watch_targets(&[first.clone(), second.clone()], &[], None, |_| false); assert_eq!(targets.len(), 2); assert!(targets.iter().any(|target| target.branch == "feature/one")); assert!(targets.iter().any(|target| target.branch == "feature/two")); @@ -512,7 +525,7 @@ mod tests { let status = snapshot("local", "/repo", "checkout"); assert!(change_request_for_chat(&chat, &[], [&status]).is_none()); assert!(conversation_branch(&chat, &[]).is_none()); - assert!(desired_watch_targets(&[chat], &[], |_| false).is_empty()); + assert!(desired_watch_targets(&[chat], &[], None, |_| false).is_empty()); } #[test] @@ -520,7 +533,7 @@ mod tests { let chat = chat("chat", "local", None, None); let status = snapshot("local", "/project", "checkout"); assert!(change_request_for_chat(&chat, &[space("local")], [&status]).is_none()); - assert!(desired_watch_targets(&[chat], &[space("local")], |_| false).is_empty()); + assert!(desired_watch_targets(&[chat], &[space("local")], None, |_| false).is_empty()); } #[test] @@ -544,16 +557,16 @@ mod tests { chat("two", "local", Some("/repo"), Some("checkout")), "feature/pr", ); - let targets = desired_watch_targets(&[first.clone(), second.clone()], &[], |_| false); + let targets = desired_watch_targets(&[first.clone(), second.clone()], &[], None, |_| false); assert_eq!(targets.len(), 1); second.archived = true; - let targets = desired_watch_targets(&[first.clone(), second.clone()], &[], |_| false); + let targets = desired_watch_targets(&[first.clone(), second.clone()], &[], None, |_| false); assert_eq!(targets.len(), 1); let mut first = first; first.archived = true; - assert!(desired_watch_targets(&[first, second], &[], |_| false).is_empty()); + assert!(desired_watch_targets(&[first, second], &[], None, |_| false).is_empty()); } #[test] @@ -576,7 +589,8 @@ mod tests { assert!(state.change_request_for_chat(&chat, &[]).is_none()); assert!( - desired_watch_targets(&[chat], &[], |device| !state.is_supported(device)).is_empty() + desired_watch_targets(&[chat], &[], None, |device| !state.is_supported(device)) + .is_empty() ); } @@ -594,7 +608,7 @@ mod tests { assert!(state.clear_unsupported_on_version_change("host", Some("0.2.3"))); assert!(state.is_supported("host")); assert_eq!( - desired_watch_targets(&[chat], &[], |device| !state.is_supported(device)).len(), + desired_watch_targets(&[chat], &[], None, |device| !state.is_supported(device)).len(), 1 ); } diff --git a/crates/ui/src/state.rs b/crates/ui/src/state.rs index ae06385237..23cc32e3e8 100644 --- a/crates/ui/src/state.rs +++ b/crates/ui/src/state.rs @@ -2583,9 +2583,12 @@ impl AppState { return; }; let targets = if self.change_requests_visible { - desired_watch_targets(&self.chats, &self.spaces, |device| { - !self.change_requests.is_supported(device) - }) + desired_watch_targets( + &self.chats, + &self.spaces, + self.selected_chat.as_deref(), + |device| !self.change_requests.is_supported(device), + ) } else { HashSet::new() }; @@ -2792,6 +2795,9 @@ impl AppState { self.start_chat_watches(chat_id, cx); } } + // A selected side chat (and a selected parent's side chats) shows + // PR badges, so the watched set follows the selection. + self.reconcile_change_request_watches(cx); cx.notify(); } @@ -5208,6 +5214,77 @@ mod tests { ); } + fn watched_branches(state: &AppState) -> Vec<&str> { + let mut branches: Vec<&str> = state + .change_request_tasks + .keys() + .map(|target| target.branch.as_str()) + .collect(); + branches.sort_unstable(); + branches + } + + /// Every watch task is a host-side Git inspection loop, so the desktop + /// holds none while PR badges are hidden, and otherwise only for chats + /// that can show a badge: top-level rows, plus the selected chat and its + /// side chats (the composer footer and the child-session rows). + #[gpui::test] + fn change_request_watches_follow_visibility_and_badge_surfaces(cx: &mut gpui::TestAppContext) { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let _guard = runtime.enter(); + let (out, _requests) = tokio::sync::mpsc::channel(256); + let (_replies, inbound) = tokio::sync::mpsc::channel(16); + let engine = EngineHandle::from_test_client(roboco_rpc::RpcClient::new(out, inbound)); + let with_branch = |mut chat: Chat, branch: &str, parent: Option<&str>| { + chat.parent_chat_id = parent.map(str::to_owned); + chat.source_context = Some(roboco_proto::ConversationSourceContext { + checkout_id: format!("checkout-{branch}"), + repo_root: format!("/worktrees/{branch}"), + cwd: format!("/worktrees/{branch}"), + branch: branch.into(), + head_sha: None, + observed_at: chat.created_at, + }); + chat + }; + let mut archived = with_branch(chat("archived", 4, None), "archived", None); + archived.archived = true; + let chats = vec![ + with_branch(chat("lead", 0, None), "lead", None), + with_branch(chat("lead-side", 1, None), "lead-side", Some("lead")), + with_branch(chat("other", 2, None), "other", None), + with_branch(chat("other-side", 3, None), "other-side", Some("other")), + archived, + ]; + + let state = cx.new(|_| AppState::new()); + state.update(cx, |state, cx| { + // The shell applies the persisted "hide PR badges" setting when it + // is built, before the engine registry attaches. + state.set_change_requests_visible(false, cx); + state.set_test_engine(engine); + state.apply_chats(chats); + state.reconcile_change_request_watches(cx); + assert!(watched_branches(state).is_empty()); + state.select_chat(Some("lead".into()), cx); + assert!(watched_branches(state).is_empty()); + state.select_chat(None, cx); + + state.set_change_requests_visible(true, cx); + assert_eq!(watched_branches(state), ["lead", "other"]); + state.select_chat(Some("lead".into()), cx); + assert_eq!(watched_branches(state), ["lead", "lead-side", "other"]); + state.select_chat(Some("other-side".into()), cx); + assert_eq!(watched_branches(state), ["lead", "other", "other-side"]); + + state.set_change_requests_visible(false, cx); + assert!(watched_branches(state).is_empty()); + }); + } + #[test] fn pending_side_chat_waits_for_registry_then_observes_deletion() { let mut state = AppState::new();