hatchery_node/
git_watch.rs

1//! Git state of live sessions for the task manager: branch, head and dirty count of the
2//! directory each session's process runs in. Read with `git` itself (no parsing of `.git`);
3//! a session outside a repository reports nothing. Unix reads the cwd from `/proc/<pid>/cwd`;
4//! elsewhere the watcher reports nothing until a portable cwd source exists.
5
6use serde::Serialize;
7use std::collections::BTreeMap;
8use std::path::{Path, PathBuf};
9use std::sync::{Mutex, OnceLock};
10
11#[derive(Clone, Debug, Serialize, PartialEq, Eq)]
12pub struct SessionGit {
13    pub workspace_id: String,
14    pub instance_id: u64,
15    pub provider: String,
16    pub branch: String,
17    pub head: String,
18    /// Changed + untracked paths (`git status --porcelain` lines), capped at 10000.
19    pub dirty: u32,
20    /// Commits on HEAD that the local default branch (`main`, else `master`) does not have.
21    pub ahead: u32,
22    pub root: String,
23    /// The m4a nick of the session that owns this worktree: its identity in core.
24    #[serde(skip_serializing_if = "Option::is_none")]
25    pub nick: Option<String>,
26    /// That session's m4a public key (Ed25519, base64), once its account store is open.
27    #[serde(skip_serializing_if = "Option::is_none")]
28    pub m4a_key: Option<String>,
29}
30
31/// Instance -> m4a nick of the node's sessions (the node assigned it at spawn).
32static NICKS: OnceLock<Mutex<BTreeMap<u64, String>>> = OnceLock::new();
33/// Nick -> its m4a public key (base64; public only).
34static KEYS: OnceLock<Mutex<BTreeMap<String, String>>> = OnceLock::new();
35
36fn nicks() -> std::sync::MutexGuard<'static, BTreeMap<u64, String>> {
37    NICKS.get_or_init(Default::default).lock().unwrap_or_else(|p| p.into_inner())
38}
39
40fn keys() -> std::sync::MutexGuard<'static, BTreeMap<String, String>> {
41    KEYS.get_or_init(Default::default).lock().unwrap_or_else(|p| p.into_inner())
42}
43
44/// The m4a nick the node gave session `instance`.
45pub fn bind_nick(instance: u64, nick: &str) {
46    nicks().insert(instance, nick.to_owned());
47}
48
49pub fn nick_of(instance: u64) -> Option<String> {
50    nicks().get(&instance).cloned()
51}
52
53pub fn m4a_key_of(nick: &str) -> Option<String> {
54    keys().get(nick).cloned()
55}
56
57/// A binding core should learn (`task.session_bind`): nick, m4a public key, and where the
58/// session lives when the node knows it.
59#[derive(Clone, Debug, PartialEq, Eq)]
60pub struct NickBinding {
61    pub nick: String,
62    pub m4a_key: String,
63    pub workspace: Option<String>,
64    pub instance: Option<u64>,
65}
66
67static BINDINGS: OnceLock<tokio::sync::mpsc::UnboundedSender<NickBinding>> = OnceLock::new();
68
69pub fn set_binding_sink(tx: tokio::sync::mpsc::UnboundedSender<NickBinding>) {
70    let _ = BINDINGS.set(tx);
71}
72
73/// The session account of `nick` is open and its public key is `key` (base64): remember it
74/// and tell core (on change).
75pub fn session_m4a_key(nick: &str, key: &str) {
76    if key.is_empty() || key.len() > 128 {
77        return;
78    }
79    let changed = keys().insert(nick.to_owned(), key.to_owned()).as_deref() != Some(key);
80    let instance = nicks().iter().find(|(_, n)| n.as_str() == nick).map(|(i, _)| *i);
81    let workspace = instance.and_then(|i| live().keys().find(|(_, x)| *x == i).map(|(w, _)| w.clone()));
82    if changed {
83        if let Some(tx) = BINDINGS.get() {
84            let _ = tx.send(NickBinding { nick: nick.to_owned(), m4a_key: key.to_owned(), workspace, instance });
85        }
86    }
87}
88
89#[derive(Clone, Debug)]
90struct Live {
91    pid: u32,
92    provider: String,
93    /// Where the session was last seen working (start cwd, then the last snapshot's root):
94    /// after exit `/proc/<pid>/cwd` is gone, and the final git state is read from here.
95    dir: Option<PathBuf>,
96}
97
98/// A session's lifecycle as the task manager sees it (`task.session_state`).
99#[derive(Clone, Debug, PartialEq, Eq)]
100pub struct Lifecycle {
101    pub workspace: String,
102    pub instance: u64,
103    /// `live` | `gone`.
104    pub state: &'static str,
105    /// For `gone`: where to read the session's last git state.
106    pub dir: Option<PathBuf>,
107    /// The session's keystore public key (hex), once it has one: core binds the session
108    /// record to it, so the session's own signed task acts name it.
109    pub public_key: Option<String>,
110}
111
112static LIFECYCLE: OnceLock<tokio::sync::mpsc::UnboundedSender<Lifecycle>> = OnceLock::new();
113
114/// Install the lifecycle sink (once, at startup). Without one, nothing is sent.
115pub fn set_lifecycle_sink(tx: tokio::sync::mpsc::UnboundedSender<Lifecycle>) {
116    let _ = LIFECYCLE.set(tx);
117}
118
119fn emit(l: Lifecycle) {
120    if let Some(tx) = LIFECYCLE.get() {
121        let _ = tx.send(l);
122    }
123}
124
125static LIVE: OnceLock<Mutex<BTreeMap<(String, u64), Live>>> = OnceLock::new();
126
127fn live() -> std::sync::MutexGuard<'static, BTreeMap<(String, u64), Live>> {
128    LIVE.get_or_init(Default::default).lock().unwrap_or_else(|p| p.into_inner())
129}
130
131/// The session's process started.
132pub fn running(workspace: &str, instance: u64, provider: &str, pid: u32) {
133    live().insert((workspace.to_owned(), instance), Live { pid, provider: provider.to_owned(), dir: cwd_of(pid) });
134    emit(Lifecycle { workspace: workspace.to_owned(), instance, state: "live", dir: None, public_key: None });
135}
136
137/// The provider of a live session (from the frame publisher; `running` may not know it).
138pub fn provider_hint(workspace: &str, instance: u64, provider: &str) {
139    if let Some(l) = live().get_mut(&(workspace.to_owned(), instance)) {
140        if l.provider != provider {
141            l.provider = provider.to_owned();
142        }
143    }
144}
145
146/// The session's process exited.
147pub fn exited(workspace: &str, instance: u64) {
148    if let Some(l) = live().remove(&(workspace.to_owned(), instance)) {
149        emit(Lifecycle { workspace: workspace.to_owned(), instance, state: "gone", dir: l.dir, public_key: None });
150    }
151}
152
153/// The keystore issued the identity of proxy session `<incarnation>-<instance>`: report the
154/// live session again with its public key.
155pub fn session_key(proxy_session: &str, public_key_hex: &str) {
156    let Some(instance) = proxy_session.rsplit('-').next().and_then(|i| i.parse::<u64>().ok()) else { return };
157    if public_key_hex.is_empty() {
158        return;
159    }
160    let workspace = live().keys().find(|(_, i)| *i == instance).map(|(w, _)| w.clone());
161    if let Some(workspace) = workspace {
162        emit(Lifecycle { workspace, instance, state: "live", dir: None, public_key: Some(public_key_hex.to_owned()) });
163    }
164}
165
166/// The git state of the repository a session worked in (`dir` itself, or the `task-*`
167/// repository one level below it), in the `task.session_git` shape.
168pub fn read_session_repo(dir: &Path) -> Option<serde_json::Value> {
169    let (branch, head, dirty, ahead, root) = read_repo_full(dir).or_else(|| task_repo_below(dir))?;
170    Some(serde_json::json!({ "branch": branch, "head": head, "dirty": dirty, "ahead": ahead, "root": root }))
171}
172
173fn cwd_of(pid: u32) -> Option<PathBuf> {
174    if cfg!(unix) {
175        std::fs::read_link(format!("/proc/{pid}/cwd")).ok()
176    } else {
177        None
178    }
179}
180
181fn git(dir: &Path, args: &[&str]) -> Option<String> {
182    let mut cmd = std::process::Command::new("git");
183    cmd.arg("-C").arg(dir).args(args).stdin(std::process::Stdio::null()).stderr(std::process::Stdio::null());
184    crate::windows_process::hide_console(&mut cmd);
185    let out = cmd.output().ok()?;
186    out.status.success().then(|| String::from_utf8_lossy(&out.stdout).into_owned())
187}
188
189/// Branch / head / dirty of the repository containing `dir`.
190pub fn read_repo(dir: &Path) -> Option<(String, String, u32, String)> {
191    read_repo_full(dir).map(|(b, h, d, _, r)| (b, h, d, r))
192}
193
194/// Like [`read_repo`] plus `ahead` of the default branch.
195pub fn read_repo_full(dir: &Path) -> Option<(String, String, u32, u32, String)> {
196    let (branch, head, dirty, root) = read_repo_basic(dir)?;
197    let base = ["main", "master"].into_iter().find(|b| git(dir, &["rev-parse", "--verify", "--quiet", &format!("refs/heads/{b}")]).is_some());
198    let ahead = base
199        .filter(|b| *b != branch)
200        .and_then(|b| git(dir, &["rev-list", "--count", "HEAD", &format!("^{b}")]))
201        .and_then(|s| s.trim().parse().ok())
202        .unwrap_or(0);
203    Some((branch, head, dirty, ahead, root))
204}
205
206fn read_repo_basic(dir: &Path) -> Option<(String, String, u32, String)> {
207    let root = git(dir, &["rev-parse", "--show-toplevel"])?.trim().to_owned();
208    let branch = git(dir, &["rev-parse", "--abbrev-ref", "HEAD"]).map(|s| s.trim().to_owned()).unwrap_or_default();
209    let head = git(dir, &["rev-parse", "--short=12", "HEAD"]).map(|s| s.trim().to_owned()).unwrap_or_default();
210    let dirty = git(dir, &["status", "--porcelain", "--untracked-files=normal"]).map(|s| s.lines().take(10_000).count() as u32).unwrap_or(0);
211    Some((branch, head, dirty, root))
212}
213
214/// A session whose cwd is not a repository (a shared workspace root): the repository one level
215/// below it that sits on a `task-*` branch (the branch a dispatched task works on).
216fn task_repo_below(dir: &Path) -> Option<(String, String, u32, u32, String)> {
217    let mut entries: Vec<_> = std::fs::read_dir(dir).ok()?.flatten().take(256).map(|e| e.path()).filter(|p| p.join(".git").exists()).collect();
218    entries.sort();
219    entries.into_iter().find_map(|p| read_repo_full(&p).filter(|r| r.0.starts_with("task-")))
220}
221
222/// One snapshot of every live session that sits in a repository.
223pub fn snapshot() -> Vec<SessionGit> {
224    let sessions: Vec<_> = live().iter().map(|(k, v)| (k.clone(), v.clone())).collect();
225    let out: Vec<SessionGit> = sessions
226        .into_iter()
227        .filter_map(|((ws, inst), l)| {
228            let dir = cwd_of(l.pid)?;
229            let (branch, head, dirty, ahead, root) = read_repo_full(&dir).or_else(|| task_repo_below(&dir))?;
230            let nick = nick_of(inst);
231            let m4a_key = nick.as_deref().and_then(m4a_key_of);
232            Some(SessionGit { workspace_id: ws, instance_id: inst, provider: l.provider, branch, head, dirty, ahead, root, nick, m4a_key })
233        })
234        .collect();
235    let mut live = live();
236    for s in &out {
237        if let Some(l) = live.get_mut(&(s.workspace_id.clone(), s.instance_id)) {
238            l.dir = Some(PathBuf::from(&s.root));
239        }
240    }
241    out
242}
243
244#[cfg(test)]
245mod tests {
246    use super::*;
247
248    #[test]
249    fn reads_branch_and_dirty() {
250        let d = std::env::temp_dir().join(format!("gw-{}", std::process::id()));
251        let _ = std::fs::remove_dir_all(&d);
252        std::fs::create_dir_all(&d).unwrap();
253        let run = |a: &[&str]| assert!(std::process::Command::new("git").arg("-C").arg(&d).args(a).output().unwrap().status.success());
254        run(&["init", "-q", "-b", "feat-x"]);
255        run(&["-c", "core.hooksPath=/dev/null", "-c", "user.name=t", "-c", "user.email=t@t", "commit", "-q", "--allow-empty", "-m", "i"]);
256        std::fs::write(d.join("a.txt"), "x").unwrap();
257        let (branch, head, dirty, _) = read_repo(&d).unwrap();
258        run(&["branch", "main"]);
259        run(&["-c", "core.hooksPath=/dev/null", "-c", "user.name=t", "-c", "user.email=t@t", "commit", "-q", "--allow-empty", "-m", "j"]);
260        assert_eq!(read_repo_full(&d).unwrap().3, 1);
261        assert_eq!(branch, "feat-x");
262        assert_eq!(head.len(), 12);
263        assert_eq!(dirty, 1);
264        let _ = std::fs::remove_dir_all(d);
265    }
266
267    #[test]
268    fn finds_task_repo_below_workspace() {
269        let ws = std::env::temp_dir().join(format!("gw-ws-{}", std::process::id()));
270        let _ = std::fs::remove_dir_all(&ws);
271        let r = ws.join("repo");
272        std::fs::create_dir_all(&r).unwrap();
273        let run = |a: &[&str]| assert!(std::process::Command::new("git").arg("-C").arg(&r).args(a).output().unwrap().status.success());
274        run(&["init", "-q", "-b", "main"]);
275        run(&["-c", "core.hooksPath=/dev/null", "-c", "user.name=t", "-c", "user.email=t@t", "commit", "-q", "--allow-empty", "-m", "i"]);
276        assert!(task_repo_below(&ws).is_none());
277        run(&["checkout", "-q", "-b", "task-9"]);
278        run(&["-c", "core.hooksPath=/dev/null", "-c", "user.name=t", "-c", "user.email=t@t", "commit", "-q", "--allow-empty", "-m", "w"]);
279        let f = task_repo_below(&ws).unwrap();
280        assert_eq!((f.0.as_str(), f.2, f.3), ("task-9", 0, 1));
281        let _ = std::fs::remove_dir_all(ws);
282    }
283
284    #[test]
285    fn lifecycle_is_emitted_with_the_last_dir() {
286        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
287        set_lifecycle_sink(tx);
288        running("lc", 91, "codex", std::process::id());
289        session_key("inc-1-91", "ab");
290        exited("lc", 91);
291        exited("lc", 91);
292        let mut got = Vec::new();
293        while let Ok(l) = rx.try_recv() {
294            if l.workspace == "lc" {
295                got.push(l);
296            }
297        }
298        assert_eq!(got.iter().map(|l| l.state).collect::<Vec<_>>(), ["live", "live", "gone"], "a second exit is not a second gone");
299        assert_eq!(got[1].public_key.as_deref(), Some("ab"));
300        if cfg!(target_os = "linux") {
301            assert_eq!(got[2].dir, std::env::current_dir().ok());
302        }
303        let here = std::env::current_dir().unwrap();
304        assert_eq!(read_session_repo(&here).is_some(), read_repo(&here).is_some());
305    }
306
307    #[test]
308    fn live_session_snapshot() {
309        running("w", 77, "claude", std::process::id());
310        let here = std::env::current_dir().unwrap();
311        let in_repo = read_repo(&here).is_some();
312        assert_eq!(snapshot().iter().any(|s| s.instance_id == 77), in_repo && cfg!(target_os = "linux"));
313        exited("w", 77);
314        assert!(!snapshot().iter().any(|s| s.instance_id == 77));
315    }
316}
317
318#[cfg(test)]
319mod nick_tests {
320    use super::*;
321
322    #[test]
323    fn nick_key_binding_goes_out_once_per_key() {
324        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
325        set_binding_sink(tx);
326        bind_nick(90_001, "claude-opus-ab12.box.lavr");
327        assert_eq!(nick_of(90_001).as_deref(), Some("claude-opus-ab12.box.lavr"));
328        session_m4a_key("claude-opus-ab12.box.lavr", "AAAA");
329        session_m4a_key("claude-opus-ab12.box.lavr", "AAAA");
330        let b = rx.try_recv().expect("binding");
331        assert_eq!((b.nick.as_str(), b.m4a_key.as_str(), b.instance), ("claude-opus-ab12.box.lavr", "AAAA", Some(90_001)));
332        assert!(rx.try_recv().is_err(), "an unchanged key is not resent");
333        assert_eq!(m4a_key_of("claude-opus-ab12.box.lavr").as_deref(), Some("AAAA"));
334    }
335}
hatchery_node/git_watch.rs

hatchery_node/
git_watch.rs
