agentmux_srv/backend/work_facts/
git.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! The git half of an agent's work facts: repository, branch and
5//! uncommitted files in its working folder.
6//!
7//! git never runs on a request: a background loop probes each agent's folder
8//! and the request reads the cache. An entry is re-probed once it is
9//! [`REFRESH_AFTER_MS`] old or the agent's folder changed. Distinct folders
10//! are probed concurrently ([`MAX_CONCURRENT_PROBES`] at a time) and a probe
11//! is abandoned after [`PROBE_TIMEOUT`], so with a handful of folders an
12//! entry is under ~30 s old; many slow repositories can make it older, which
13//! is why every answer carries `git_checked_ms`. Every git run goes through
14//! `fs_ops::git`'s safe runner (the repository's own fsmonitor and filters
15//! are off, nothing prompts, the index isn't written, a slow run is killed
16//! after a few seconds).
17
18use std::collections::{HashMap, HashSet};
19use std::future::Future;
20use std::sync::Arc;
21use std::time::Duration;
22
23use futures_util::StreamExt;
24
25use serde::{Deserialize, Serialize};
26
27use crate::backend::fs_ops::git::{parse_porcelain_v2, safe_git_line, safe_status_porcelain};
28use crate::backend::rpc_types::FsGitState;
29use crate::backend::storage::store::Store;
30
31use super::paths::{normalize_path, normalize_repo};
32
33/// How often the loop looks for entries to refresh.
34const TICK: Duration = Duration::from_secs(5);
35
36/// An entry this old is probed again on the next tick.
37pub const REFRESH_AFTER_MS: u64 = 20_000;
38
39/// Most probes running at once: each is a few git processes.
40const MAX_CONCURRENT_PROBES: usize = 4;
41
42/// A probe (several git runs, each with its own 4 s limit) is abandoned
43/// after this and the folder reads as having no git facts until the next.
44const PROBE_TIMEOUT: Duration = Duration::from_secs(8);
45
46/// Most uncommitted paths kept per agent; [`GitFacts::dirty_count`] is the
47/// real total.
48const MAX_DIRTY_FILES: usize = 500;
49
50/// What git says about one working folder.
51#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
52pub struct GitFacts {
53    /// The repository's top folder, normalized.
54    pub root: String,
55    /// `owner/repo` from the `origin` remote; `None` without one.
56    pub repo: Option<String>,
57    /// `None` when detached or unknown.
58    pub branch: Option<String>,
59    /// Repository-relative paths with uncommitted changes (staged, unstaged,
60    /// untracked; an untracked folder is reported whole, ending in `/`).
61    pub dirty_files: Vec<String>,
62    pub dirty_count: usize,
63    /// Why the status part is missing, when it is.
64    #[serde(default, skip_serializing_if = "Option::is_none")]
65    pub error: Option<String>,
66    pub checked_ms: u64,
67}
68
69/// Probe `dir`: `None` when it isn't inside a git repository (or git is
70/// missing or too slow).
71pub async fn probe(dir: &str, now_ms: u64) -> Option<GitFacts> {
72    let root = normalize_path(&safe_git_line(dir, &["rev-parse", "--show-toplevel"]).await?);
73    let repo = safe_git_line(&root, &["remote", "get-url", "origin"])
74        .await
75        .and_then(|url| normalize_repo(&url));
76    let mut facts = GitFacts { root, repo, checked_ms: now_ms, ..Default::default() };
77    match safe_status_porcelain(&facts.root, &[]).await {
78        Ok(out) => {
79            let p = parse_porcelain_v2(&out);
80            facts.branch = p.branch;
81            let dirty: Vec<String> =
82                p.records.into_iter().filter(|r| r.state != FsGitState::Ignored).map(|r| r.path).collect();
83            facts.dirty_count = dirty.len();
84            facts.dirty_files = dirty.into_iter().take(MAX_DIRTY_FILES).collect();
85        }
86        Err(e) => {
87            facts.branch = safe_git_line(&facts.root, &["branch", "--show-current"]).await;
88            facts.error = Some(match e {
89                crate::backend::fs_ops::git::StatusError::Settings(what) => what.to_string(),
90                crate::backend::fs_ops::git::StatusError::Git(m) => m,
91            });
92        }
93    }
94    Some(facts)
95}
96
97#[derive(Debug, Clone)]
98struct Entry {
99    dir: String,
100    facts: Option<GitFacts>,
101    checked_ms: u64,
102}
103
104/// Probe results per block id, with the folder each was taken in.
105#[derive(Debug, Default)]
106pub struct GitCache {
107    entries: HashMap<String, Entry>,
108}
109
110impl GitCache {
111    /// Whether `block`'s folder `dir` should be probed now.
112    pub fn needs_probe(&self, block: &str, dir: &str, now_ms: u64) -> bool {
113        match self.entries.get(block) {
114            None => true,
115            Some(e) => e.dir != dir || now_ms.saturating_sub(e.checked_ms) >= REFRESH_AFTER_MS,
116        }
117    }
118
119    pub fn put(&mut self, block: &str, dir: &str, facts: Option<GitFacts>, now_ms: u64) {
120        self.entries.insert(block.to_string(), Entry { dir: dir.to_string(), facts, checked_ms: now_ms });
121    }
122
123    /// The facts for `block`, only if they were taken in `dir`: after a `cd`
124    /// the old repository is no longer what the agent is working in.
125    pub fn get(&self, block: &str, dir: &str) -> Option<GitFacts> {
126        self.entries.get(block).filter(|e| e.dir == dir).and_then(|e| e.facts.clone())
127    }
128
129    pub fn retain(&mut self, live: &HashSet<String>) {
130        self.entries.retain(|b, _| live.contains(b));
131    }
132}
133
134fn cache() -> &'static parking_lot::RwLock<GitCache> {
135    static CACHE: std::sync::LazyLock<parking_lot::RwLock<GitCache>> = std::sync::LazyLock::new(Default::default);
136    &CACHE
137}
138
139/// The cached facts for `block` in `dir` (see [`GitCache::get`]).
140pub fn cached(block: &str, dir: &str) -> Option<GitFacts> {
141    cache().read().get(block, dir)
142}
143
144/// One refresh pass: probe each due `(block, folder)` in `folders`, every
145/// distinct folder once and at most [`MAX_CONCURRENT_PROBES`] at a time,
146/// each abandoned after `timeout`, and store the results stamped `now_ms`.
147pub async fn refresh_once<F, Fut>(
148    cache: &parking_lot::RwLock<GitCache>,
149    folders: Vec<(String, String)>,
150    now_ms: u64,
151    timeout: Duration,
152    probe: F,
153) where
154    F: Fn(String) -> Fut,
155    Fut: Future<Output = Option<GitFacts>>,
156{
157    let due: Vec<(String, String)> = {
158        let c = cache.read();
159        folders.into_iter().filter(|(b, d)| c.needs_probe(b, d, now_ms)).collect()
160    };
161    let mut dirs: Vec<String> = due.iter().map(|(_, d)| d.clone()).collect();
162    dirs.sort();
163    dirs.dedup();
164    let results: HashMap<String, Option<GitFacts>> = futures_util::stream::iter(dirs)
165        .map(|dir| {
166            let fut = probe(dir.clone());
167            async move { (dir, tokio::time::timeout(timeout, fut).await.ok().flatten()) }
168        })
169        .buffer_unordered(MAX_CONCURRENT_PROBES)
170        .collect()
171        .await;
172    let mut c = cache.write();
173    for (block, dir) in due {
174        c.put(&block, &dir, results.get(&dir).cloned().flatten(), now_ms);
175    }
176}
177
178/// Keep every registered agent's git facts fresh. Never returns.
179pub async fn run_git_refresh_loop(mstore: Arc<Store>) {
180    let mut ticker = tokio::time::interval(TICK);
181    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
182    loop {
183        ticker.tick().await;
184        let agents = crate::backend::reactive::get_global_handler().list_agents();
185        let live: HashSet<String> = agents.iter().map(|a| a.block_id.clone()).collect();
186        cache().write().retain(&live);
187
188        // Block meta lives in the store: read it off the async workers.
189        let store = mstore.clone();
190        let Ok(folders) = tokio::task::spawn_blocking(move || {
191            let home = dirs::home_dir();
192            agents
193                .iter()
194                .filter_map(|a| super::folder_of(&store, home.as_deref(), a).map(|d| (a.block_id.clone(), d)))
195                .collect::<Vec<_>>()
196        })
197        .await
198        else {
199            continue;
200        };
201
202        let now = agentmux_common::time::now_ms_u64();
203        refresh_once(cache(), folders, now, PROBE_TIMEOUT, |dir| async move { probe(&dir, now).await }).await;
204    }
205}
206
207/// A branch name taken from a command, safe to hand git: no option, range or
208/// whitespace.
209fn plain_branch(name: &str) -> bool {
210    !name.is_empty() && !name.starts_with('-') && !name.contains("..") && !name.contains(char::is_whitespace)
211}
212
213/// The first of `candidates` that names a commit in `root`.
214async fn first_ref(root: &str, candidates: &[String]) -> Option<String> {
215    for c in candidates {
216        if safe_git_line(root, &["rev-parse", "--verify", "--quiet", &format!("{c}^{{commit}}")]).await.is_some() {
217            return Some(c.clone());
218        }
219    }
220    None
221}
222
223/// The branch's `gh-merge-base` setting (gh's base for a PR from it), if set.
224pub async fn merge_base_setting(root: &str, branch: &str) -> Option<String> {
225    if !plain_branch(branch) {
226        return None;
227    }
228    safe_git_line(root, &["config", "--get", &format!("branch.{branch}.gh-merge-base")]).await.filter(|b| plain_branch(b))
229}
230
231/// The files a PR from `head` into `base` changes in `root`: `git diff
232/// --name-only base...head`, the base being `origin/<base>` (or a local
233/// `<base>`), else the default branch (`origin/HEAD`, `origin/main`,
234/// `origin/master`), and the head a local branch, else `origin/<head>`. Empty
235/// when either can't be found, on the base itself, or when git can't say in
236/// [`PROBE_TIMEOUT`]. A diff of two commits: no working-tree filter runs.
237pub async fn branch_files(root: &str, base: Option<&str>, head: &str) -> Vec<String> {
238    if !plain_branch(head) || base.is_some_and(|b| !plain_branch(b)) {
239        return Vec::new();
240    }
241    let files = async {
242        let bases: Vec<String> = match base {
243            Some(b) => vec![format!("origin/{b}"), b.to_string()],
244            None => ["origin/HEAD", "origin/main", "origin/master"].iter().map(|s| s.to_string()).collect(),
245        };
246        let base = first_ref(root, &bases).await?;
247        let head = first_ref(root, &[format!("refs/heads/{head}"), format!("origin/{head}")]).await?;
248        let out = safe_git_line(root, &["diff", "--name-only", &format!("{base}...{head}")]).await?;
249        Some(out.lines().map(str::trim).filter(|l| !l.is_empty()).map(str::to_string).collect::<Vec<_>>())
250    };
251    tokio::time::timeout(PROBE_TIMEOUT, files).await.ok().flatten().unwrap_or_default()
252}
253
254#[cfg(test)]
255mod tests {
256    use super::*;
257
258    fn git(dir: &std::path::Path, args: &[&str]) {
259        use agentmux_common::win32::NoWindow;
260        let ok = std::process::Command::new("git")
261            .args(args)
262            .current_dir(dir)
263            .env("GIT_AUTHOR_NAME", "t")
264            .env("GIT_AUTHOR_EMAIL", "t@t")
265            .env("GIT_COMMITTER_NAME", "t")
266            .env("GIT_COMMITTER_EMAIL", "t@t")
267            .no_window()
268            .output()
269            .map(|o| o.status.success())
270            .unwrap_or(false);
271        assert!(ok, "git {args:?} failed");
272    }
273
274    /// A temp repository gives its repository, branch and uncommitted files.
275    #[tokio::test]
276    async fn a_temp_repo_gives_repo_branch_and_dirty_files() {
277        let tmp = tempfile::tempdir().unwrap();
278        let dir = tmp.path();
279        git(dir, &["init", "-q", "-b", "feat/x"]);
280        git(dir, &["remote", "add", "origin", "https://github.com/AgentMuxAI/Example.git"]);
281        std::fs::create_dir_all(dir.join("src")).unwrap();
282        std::fs::write(dir.join("src/a.rs"), "a").unwrap();
283        git(dir, &["add", "."]);
284        git(dir, &["-c", "commit.gpgsign=false", "commit", "-q", "-m", "init"]);
285        std::fs::write(dir.join("src/a.rs"), "changed").unwrap();
286        std::fs::create_dir_all(dir.join("new")).unwrap();
287        std::fs::write(dir.join("new/b.rs"), "b").unwrap();
288
289        let sub = dir.join("src");
290        let facts = probe(&sub.to_string_lossy(), 42).await.expect("inside a repository");
291        assert_eq!(facts.repo.as_deref(), Some("agentmuxai/example"));
292        assert_eq!(facts.branch.as_deref(), Some("feat/x"));
293        let mut dirty = facts.dirty_files.clone();
294        dirty.sort();
295        assert_eq!(dirty, vec!["new/".to_string(), "src/a.rs".to_string()]);
296        assert_eq!(facts.dirty_count, 2);
297        assert_eq!(facts.checked_ms, 42);
298        let name = dir.file_name().unwrap().to_string_lossy().into_owned();
299        assert!(facts.root.ends_with(&name), "root {} is the temp repository, not its src folder", facts.root);
300    }
301
302    #[tokio::test]
303    async fn a_folder_outside_any_repository_has_no_git_facts() {
304        let tmp = tempfile::tempdir().unwrap();
305        // A temp folder might sit inside a repository on a dev machine; only
306        // assert when git agrees it doesn't.
307        if let Some(f) = probe(&tmp.path().to_string_lossy(), 1).await {
308            assert!(!f.root.is_empty());
309        }
310    }
311
312    /// The 30 s cache: an entry is reused until it is REFRESH_AFTER_MS old,
313    /// and dropped at once when the agent changes folder.
314    #[test]
315    fn the_cache_is_reused_until_stale_or_the_folder_changes() {
316        let mut c = GitCache::default();
317        assert!(c.needs_probe("b1", "/r", 0), "never probed");
318        let facts = GitFacts { root: "/r".into(), checked_ms: 1_000, ..Default::default() };
319        c.put("b1", "/r", Some(facts.clone()), 1_000);
320        assert!(!c.needs_probe("b1", "/r", 1_000 + REFRESH_AFTER_MS - 1));
321        assert!(c.needs_probe("b1", "/r", 1_000 + REFRESH_AFTER_MS));
322        assert!(c.needs_probe("b1", "/other", 1_001), "a cd re-probes at once");
323        assert_eq!(c.get("b1", "/r"), Some(facts));
324        assert_eq!(c.get("b1", "/other"), None, "facts from the old folder aren't shown");
325
326        c.put("b2", "/n", None, 1_000);
327        assert!(!c.needs_probe("b2", "/n", 2_000), "a folder outside git is cached too");
328        c.retain(&HashSet::from(["b2".to_string()]));
329        assert_eq!(c.get("b1", "/r"), None, "a gone agent's entry is dropped");
330    }
331
332    fn facts_for(dir: &str) -> GitFacts {
333        GitFacts { root: dir.to_string(), ..Default::default() }
334    }
335
336    /// The loop's real pass: distinct folders are probed concurrently (at
337    /// most MAX_CONCURRENT_PROBES), a shared folder once, a fresh entry not
338    /// at all, and a hung probe is abandoned rather than holding up the rest.
339    #[tokio::test(start_paused = true)]
340    async fn a_pass_probes_due_folders_concurrently_and_abandons_a_hung_one() {
341        use std::sync::atomic::{AtomicUsize, Ordering};
342        let cache = parking_lot::RwLock::new(GitCache::default());
343        cache.write().put("fresh", "/fresh", Some(facts_for("/fresh")), 1_000);
344
345        let mut folders: Vec<(String, String)> = (0..8).map(|i| (format!("b{i}"), format!("/r{i}"))).collect();
346        folders.push(("shares-r0".into(), "/r0".into()));
347        folders.push(("fresh".into(), "/fresh".into()));
348        folders.push(("hung".into(), "/hung".into()));
349
350        let (running, peak, calls) = (AtomicUsize::new(0), AtomicUsize::new(0), AtomicUsize::new(0));
351        let start = tokio::time::Instant::now();
352        refresh_once(&cache, folders, 2_000, Duration::from_secs(8), |dir| {
353            let (running, peak, calls) = (&running, &peak, &calls);
354            async move {
355                calls.fetch_add(1, Ordering::SeqCst);
356                let now = running.fetch_add(1, Ordering::SeqCst) + 1;
357                peak.fetch_max(now, Ordering::SeqCst);
358                let wait = if dir == "/hung" { 600 } else { 2 };
359                tokio::time::sleep(Duration::from_secs(wait)).await;
360                running.fetch_sub(1, Ordering::SeqCst);
361                Some(facts_for(&dir))
362            }
363        })
364        .await;
365
366        assert_eq!(calls.load(Ordering::SeqCst), 9, "8 folders + the hung one; shared and fresh ones skipped");
367        assert_eq!(peak.load(Ordering::SeqCst), MAX_CONCURRENT_PROBES);
368        // Eight 2 s probes four at a time beside one 8 s timeout finish in
369        // about 8 s; one at a time would take 24 s.
370        assert!(start.elapsed() <= Duration::from_secs(9), "took {:?}", start.elapsed());
371        let c = cache.read();
372        assert_eq!(c.get("b3", "/r3"), Some(facts_for("/r3")));
373        assert_eq!(c.get("shares-r0", "/r0"), Some(facts_for("/r0")));
374        assert_eq!(c.get("hung", "/hung"), None, "abandoned");
375        assert!(!c.needs_probe("hung", "/hung", 2_001), "and not retried until it is due again");
376        assert_eq!(c.get("fresh", "/fresh"), Some(facts_for("/fresh")));
377    }
378
379    #[tokio::test]
380    async fn a_branch_lists_the_files_it_changes_since_the_default_branch() {
381        let tmp = std::env::temp_dir().join(format!("wf-branch-files-{}-{}", std::process::id(), agentmux_common::time::now_ms()));
382        std::fs::create_dir_all(&tmp).unwrap();
383        git(&tmp, &["init", "-q", "-b", "main"]);
384        git(&tmp, &["config", "user.email", "t@t"]);
385        git(&tmp, &["config", "user.name", "t"]);
386        std::fs::write(tmp.join("base.rs"), "a").unwrap();
387        git(&tmp, &["add", "."]);
388        git(&tmp, &["commit", "-q", "-m", "base"]);
389        git(&tmp, &["update-ref", "refs/remotes/origin/main", "HEAD"]);
390        let root = tmp.to_string_lossy().to_string();
391        assert!(branch_files(&root, None, "main").await.is_empty(), "the default branch itself: nothing");
392        git(&tmp, &["checkout", "-q", "-b", "agent/x"]);
393        std::fs::create_dir_all(tmp.join("src")).unwrap();
394        std::fs::write(tmp.join("src/new.rs"), "b").unwrap();
395        std::fs::write(tmp.join("base.rs"), "changed").unwrap();
396        git(&tmp, &["add", "."]);
397        git(&tmp, &["commit", "-q", "-m", "work"]);
398        let both = vec!["base.rs".to_string(), "src/new.rs".to_string()];
399        assert_eq!(branch_files(&root, None, "agent/x").await, both);
400        // A PR for another head, from wherever HEAD is.
401        git(&tmp, &["checkout", "-q", "main"]);
402        assert_eq!(branch_files(&root, None, "agent/x").await, both, "--head, not the checked-out branch");
403        // A stacked PR: based on agent/x, it changes only its own commit.
404        git(&tmp, &["checkout", "-q", "-b", "agent/y", "agent/x"]);
405        std::fs::write(tmp.join("y.rs"), "y").unwrap();
406        git(&tmp, &["add", "."]);
407        git(&tmp, &["commit", "-q", "-m", "stacked"]);
408        assert_eq!(branch_files(&root, Some("agent/x"), "agent/y").await, vec!["y.rs".to_string()]);
409        git(&tmp, &["config", "branch.agent/y.gh-merge-base", "agent/x"]);
410        assert_eq!(merge_base_setting(&root, "agent/y").await.as_deref(), Some("agent/x"));
411        assert!(branch_files(&root, None, "no/such/branch").await.is_empty(), "a head that can't be found");
412        assert!(branch_files(&root, None, "--output=x").await.is_empty(), "never a flag");
413        let _ = std::fs::remove_dir_all(&tmp);
414    }
415}
agentmux_srv/backend/work_facts/git.rs

agentmux_srv/backend/work_facts/
git.rs
