1use std::path::{Path, PathBuf};
15
16use serde::{Deserialize, Serialize};
17use serde_json::Value;
18
19use crate::learn;
20
21pub const IDLE_MS: i64 = 30 * 60 * 1000;
23pub const PRUNE_MS: i64 = 24 * 60 * 60 * 1000;
25pub const FILES: usize = 3;
27const PROMPT_CHARS: usize = 100;
29
30#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
31pub struct Presence {
32 pub session: String,
33 pub harness: String,
34 #[serde(default)]
36 pub engineer: String,
37 #[serde(default)]
39 pub repo: String,
40 pub collective: String,
41 #[serde(default)]
43 pub prompt: String,
44 #[serde(default)]
46 pub files: Vec<String>,
47 pub event: String,
49 pub ms: i64,
50 pub started: i64,
52 #[serde(default, skip_serializing_if = "Option::is_none")]
54 pub pid: Option<u32>,
55}
56
57#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
58#[serde(rename_all = "lowercase")]
59pub enum State {
60 Working,
61 Waiting,
62 Idle,
63 Ended,
64}
65
66impl State {
67 pub fn name(self) -> &'static str {
68 match self {
69 State::Working => "working",
70 State::Waiting => "waiting",
71 State::Idle => "idle",
72 State::Ended => "ended",
73 }
74 }
75}
76
77pub fn dir(home: &Path) -> PathBuf {
78 home.join("presence")
79}
80
81pub fn safe(session: &str) -> String {
83 session.chars().filter(|c| c.is_ascii_alphanumeric() || *c == '-').collect()
84}
85
86pub fn path(home: &Path, session: &str) -> PathBuf {
87 dir(home).join(format!("{}.json", safe(session)))
88}
89
90pub fn remote_path(home: &Path) -> PathBuf {
92 home.join("mesh").join("presence.jsonl")
93}
94
95pub fn read(home: &Path, session: &str) -> Option<Presence> {
96 serde_json::from_slice(&std::fs::read(path(home, session)).ok()?).ok()
97}
98
99fn file_of(input: &Value, cwd: &str) -> Option<String> {
101 let ti = input.get("tool_input")?;
102 let f = ["file_path", "notebook_path"].iter().find_map(|k| ti.get(*k).and_then(Value::as_str)).filter(|f| !f.trim().is_empty())?;
103 let rel = Path::new(f).strip_prefix(cwd).ok().map(|p| p.display().to_string()).filter(|p| !p.is_empty());
104 Some(rel.unwrap_or_else(|| f.to_string()))
105}
106
107pub fn first_line(prompt: &str) -> String {
109 let line = prompt.lines().map(str::trim).find(|l| !l.is_empty()).unwrap_or_default();
110 let mut out: String = line.chars().take(PROMPT_CHARS).collect();
111 if line.chars().count() > PROMPT_CHARS {
112 out.push('…');
113 }
114 out
115}
116
117pub fn update(
120 prev: Option<Presence>,
121 collective: &str,
122 harness: &str,
123 event: &str,
124 input: &Value,
125 now: i64,
126 who: impl FnOnce(&str) -> (String, String),
127) -> Presence {
128 let cwd = input.get("cwd").and_then(Value::as_str).unwrap_or_default();
129 let session = input.get("session_id").and_then(Value::as_str).unwrap_or("unknown");
130 let mut p = prev.unwrap_or_else(|| {
131 let (engineer, repo) = who(cwd);
132 Presence {
133 session: session.into(),
134 harness: harness.into(),
135 engineer,
136 repo,
137 collective: collective.into(),
138 prompt: String::new(),
139 files: Vec::new(),
140 event: String::new(),
141 ms: now,
142 started: now,
143 pid: None,
144 }
145 });
146 p.event = event.into();
147 p.ms = now;
148 if let Some(pid) = input.get("pid").and_then(Value::as_u64).and_then(|v| u32::try_from(v).ok()) {
149 p.pid = Some(pid);
150 }
151 match event {
152 "prompt" => {
153 let line = first_line(input.get("prompt").and_then(Value::as_str).unwrap_or_default());
154 if !line.is_empty() {
155 p.prompt = line;
156 }
157 }
158 "post_tool" | "post_tool_failure" => {
159 if let Some(f) = file_of(input, cwd) {
160 p.files.retain(|x| *x != f);
161 p.files.insert(0, f);
162 p.files.truncate(FILES);
163 }
164 }
165 _ => {}
166 }
167 p
168}
169
170pub fn touch(home: &Path, collective: &str, harness: &str, event: &str, input: &Value, now: i64) -> Option<Presence> {
173 let session = input.get("session_id").and_then(Value::as_str).unwrap_or("unknown");
174 let p = update(read(home, session), collective, harness, event, input, now, |cwd| {
175 (learn::engineer(cwd), borg_trace::repo_key(cwd).rsplit('/').next().unwrap_or_default().to_string())
176 });
177 write(home, &p).ok()?;
178 if event == "prompt" {
179 prune(home, now);
180 }
181 Some(p)
182}
183
184pub fn write(home: &Path, p: &Presence) -> std::io::Result<()> {
185 let d = dir(home);
186 std::fs::create_dir_all(&d)?;
187 let target = path(home, &p.session);
188 let tmp = d.join(format!(".{}.{}.tmp", safe(&p.session), std::process::id()));
189 std::fs::write(&tmp, serde_json::to_vec(p).unwrap_or_default())?;
190 std::fs::rename(&tmp, &target).inspect_err(|_| {
191 let _ = std::fs::remove_file(&tmp);
192 })
193}
194
195pub fn prune(home: &Path, now: i64) {
197 let Ok(rd) = std::fs::read_dir(dir(home)) else { return };
198 for e in rd.flatten() {
199 let old = e
200 .metadata()
201 .and_then(|m| m.modified())
202 .ok()
203 .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
204 .is_some_and(|d| now - d.as_millis() as i64 > PRUNE_MS);
205 if old {
206 let _ = std::fs::remove_file(e.path());
209 if let Some(session) = e.path().file_stem().and_then(|s| s.to_str()).filter(|s| !s.starts_with('.')) {
211 crate::listen::forget(home, session);
212 }
213 }
214 }
215}
216
217pub fn local(home: &Path, now: i64) -> Vec<Presence> {
219 let Ok(rd) = std::fs::read_dir(dir(home)) else { return Vec::new() };
220 let mut v: Vec<Presence> = rd
221 .flatten()
222 .filter(|e| e.path().extension().is_some_and(|x| x == "json"))
223 .filter_map(|e| serde_json::from_slice::<Presence>(&std::fs::read(e.path()).ok()?).ok())
224 .filter(|p| now - p.ms <= PRUNE_MS)
225 .collect();
226 v.sort_by(|a, b| (a.started, &a.session).cmp(&(b.started, &b.session)));
227 v
228}
229
230#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
233pub struct Remote {
234 pub p: Presence,
235 pub key: String,
236 pub sig: String,
237}
238
239pub fn remote(home: &Path, now: i64) -> Vec<Remote> {
241 let text = std::fs::read_to_string(remote_path(home)).unwrap_or_default();
242 let mut v: Vec<Remote> = text.lines().filter_map(|l| serde_json::from_str::<Remote>(l).ok()).filter(|r| now - r.p.ms <= PRUNE_MS).collect();
243 v.sort_by(|a, b| (&a.key, &a.p.session, b.p.ms).cmp(&(&b.key, &b.p.session, a.p.ms)));
244 v.dedup_by(|a, b| a.key == b.key && a.p.session == b.p.session);
245 v
246}
247
248pub fn synced_path(home: &Path) -> PathBuf {
253 home.join("mesh").join("synced.json")
254}
255
256pub fn synced(home: &Path) -> std::collections::BTreeMap<String, i64> {
258 std::fs::read_to_string(synced_path(home)).ok().and_then(|t| serde_json::from_str(&t).ok()).unwrap_or_default()
259}
260
261pub fn note_synced(home: &Path, key: &str, ms: i64) -> std::io::Result<()> {
267 let mut all = synced(home);
268 if all.get(key).is_some_and(|&held| held >= ms && held <= crate::learn::now_ms()) {
269 return Ok(());
270 }
271 all.insert(key.into(), ms);
272 write_synced(home, &all)
273}
274
275pub fn forget_synced(home: &Path, key: &str) -> std::io::Result<()> {
277 let mut all = synced(home);
278 if all.remove(key).is_none() {
279 return Ok(());
280 }
281 write_synced(home, &all)
282}
283
284fn write_synced(home: &Path, all: &std::collections::BTreeMap<String, i64>) -> std::io::Result<()> {
285 write_mesh(home, &synced_path(home), all)
286}
287
288fn write_mesh<T: Serialize>(home: &Path, p: &Path, all: &T) -> std::io::Result<()> {
290 std::fs::create_dir_all(home.join("mesh"))?;
291 let tmp = p.with_extension(format!("{}.tmp", std::process::id())); std::fs::write(&tmp, serde_json::to_string(all).unwrap_or_default())?;
293 std::fs::rename(&tmp, p)
294}
295
296pub fn protocols_path(home: &Path) -> PathBuf {
301 home.join("mesh").join("protocols.json")
302}
303
304#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
306pub struct Mismatch {
307 pub theirs: String,
308 pub ours: String,
309 pub ms: i64,
310}
311
312impl Mismatch {
313 pub fn line(&self, key: &str) -> String {
315 let n = |p: &str| p.rsplit('/').next().and_then(|v| v.parse::<u32>().ok()).unwrap_or(0);
316 let who = if n(&self.theirs) < n(&self.ours) { "upgrade it" } else { "upgrade this node" };
317 format!("peer {} speaks {}, this node {}: nothing syncs with it; {who}", key.get(..12).unwrap_or(key), self.theirs, self.ours)
318 }
319
320 pub fn told(&self, me: &str) -> String {
322 Mismatch { theirs: self.ours.clone(), ours: self.theirs.clone(), ms: self.ms }.line(me)
323 }
324}
325
326pub fn mismatches(home: &Path) -> std::collections::BTreeMap<String, Mismatch> {
328 std::fs::read_to_string(protocols_path(home)).ok().and_then(|t| serde_json::from_str(&t).ok()).unwrap_or_default()
329}
330
331pub fn note_mismatch(home: &Path, key: &str, m: Mismatch) -> std::io::Result<()> {
333 let mut all = mismatches(home);
334 all.insert(key.into(), m);
335 write_mesh(home, &protocols_path(home), &all)
336}
337
338pub fn forget_mismatch(home: &Path, key: &str) -> std::io::Result<()> {
340 let mut all = mismatches(home);
341 if all.remove(key).is_none() {
342 return Ok(());
343 }
344 write_mesh(home, &protocols_path(home), &all)
345}
346
347pub fn retain_mismatches(home: &Path, keep: impl Fn(&str) -> bool) -> std::io::Result<()> {
350 let mut all = mismatches(home);
351 let before = all.len();
352 all.retain(|k, _| keep(k));
353 if all.len() == before {
354 return Ok(());
355 }
356 write_mesh(home, &protocols_path(home), &all)
357}
358
359pub fn mesh_checks(routes: &[crate::collective::Route]) -> Vec<(crate::harness::Level, String)> {
363 use crate::collective::Route;
364 routes
365 .iter()
366 .filter_map(|r| match r {
367 Route::Collective { name, home, .. } => Some((name, home)),
368 Route::Off => None,
369 })
370 .flat_map(|(name, home)| mismatches(home).into_iter().map(move |(k, m)| (crate::harness::Level::Warn, format!("mesh {name}: {}", m.line(&k)))))
371 .collect()
372}
373
374pub fn state(p: &Presence, now: i64, alive: impl Fn(u32) -> bool) -> State {
376 if p.event == "end" || p.pid.is_some_and(|pid| !alive(pid)) {
377 State::Ended
378 } else if now - p.ms > IDLE_MS {
379 State::Idle
380 } else if p.event == "stop" {
381 State::Waiting
382 } else {
383 State::Working
384 }
385}
386
387pub fn alive(pid: u32) -> bool {
389 i32::try_from(pid).is_ok_and(crate::live::alive)
390}
391
392#[cfg(test)]
393mod tests {
394 use super::*;
395 use serde_json::json;
396
397 fn tmp(tag: &str) -> PathBuf {
398 let d = std::env::temp_dir().join(format!("borg-presence-{tag}-{}", std::process::id()));
399 let _ = std::fs::remove_dir_all(&d);
400 d
401 }
402
403 fn who(_: &str) -> (String, String) {
404 ("alice@x".into(), "app".into())
405 }
406
407 #[test]
410 fn a_members_last_session_never_moves_back() {
411 let h = tmp("synced");
412 assert!(synced(&h).is_empty());
413 note_synced(&h, "k1", 200).unwrap();
414 note_synced(&h, "k1", 100).unwrap();
415 note_synced(&h, "k2", 50).unwrap();
416 assert_eq!(synced(&h).into_iter().collect::<Vec<_>>(), [("k1".to_string(), 200), ("k2".to_string(), 50)]);
417 note_synced(&h, "k1", 300).unwrap();
418 assert_eq!(synced(&h)["k1"], 300);
419 forget_synced(&h, "k1").unwrap();
420 assert_eq!(synced(&h).into_keys().collect::<Vec<_>>(), ["k2"], "a failed dial drops the member");
421 let _ = std::fs::remove_dir_all(&h);
422 }
423
424 #[test]
425 fn doctor_names_a_member_on_another_mesh_protocol() {
426 use crate::collective::Route;
427 use crate::harness::Level;
428 let home = tmp("protocols-doctor");
429 let route = Route::Collective { name: "team".into(), home: home.clone(), share: crate::collective::Share::Team };
430 assert!(mesh_checks(std::slice::from_ref(&route)).is_empty());
431 let m = Mismatch { theirs: "borg/mesh/5".into(), ours: "borg/mesh/10".into(), ms: 1 };
432 note_mismatch(&home, "abcdef0123456789", m).unwrap();
433 let checks = mesh_checks(&[route, Route::Off]);
434 assert_eq!(checks.len(), 1, "{checks:?}");
435 assert_eq!(checks[0].0, Level::Warn);
436 assert!(checks[0].1.starts_with("mesh team: peer abcdef012345 speaks borg/mesh/5") && checks[0].1.ends_with("upgrade it"), "{checks:?}");
437 }
438
439 #[test]
441 fn a_protocol_mismatch_is_kept_until_a_session_completes() {
442 let h = tmp("protocols");
443 assert!(mismatches(&h).is_empty());
444 let old = Mismatch { theirs: "borg/mesh/5".into(), ours: "borg/mesh/10".into(), ms: 1 };
445 note_mismatch(&h, "abcdef0123456789", old.clone()).unwrap();
446 note_mismatch(&h, "k2", Mismatch { theirs: "borg/mesh/11".into(), ..old.clone() }).unwrap();
447 let all = mismatches(&h);
448 assert_eq!(
449 all["abcdef0123456789"].line("abcdef0123456789"),
450 "peer abcdef012345 speaks borg/mesh/5, this node borg/mesh/10: nothing syncs with it; upgrade it"
451 );
452 assert!(all["k2"].line("k2").ends_with("upgrade this node"), "a newer peer: this node is the old one");
453 forget_mismatch(&h, "abcdef0123456789").unwrap();
454 assert_eq!(mismatches(&h).into_keys().collect::<Vec<_>>(), ["k2"]);
455 let _ = std::fs::remove_dir_all(&h);
456 }
457
458 #[test]
459 fn a_record_ahead_of_the_clock_gives_way() {
460 let h = tmp("synced-skew");
462 let now = crate::learn::now_ms();
463 note_synced(&h, "k1", now + 3_600_000).unwrap();
464 note_synced(&h, "k1", now).unwrap();
465 assert_eq!(synced(&h)["k1"], now);
466 let _ = std::fs::remove_dir_all(&h);
467 }
468
469 #[test]
470 fn a_session_record_follows_its_hooks() {
471 let ev = |e: &str, extra: Value| {
472 let mut v = json!({"session_id": "s1", "cwd": "/w/app"});
473 v.as_object_mut().unwrap().extend(extra.as_object().unwrap().clone());
474 (e.to_string(), v)
475 };
476 let mut p = None;
477 let steps = [
478 ev("prompt", json!({"prompt": "\n fix the flaky sync test\nand more", "pid": 4242})),
479 ev("post_tool", json!({"tool_name": "Edit", "tool_input": {"file_path": "/w/app/src/a.rs"}})),
480 ev("post_tool", json!({"tool_name": "Read", "tool_input": {"file_path": "/w/app/src/b.rs"}})),
481 ev("post_tool", json!({"tool_name": "Bash", "tool_input": {"command": "ls"}})),
482 ev("post_tool", json!({"tool_name": "Edit", "tool_input": {"file_path": "/w/app/src/a.rs"}})),
483 ev("post_tool", json!({"tool_name": "Write", "tool_input": {"file_path": "/elsewhere/c.rs"}})),
484 ev("post_tool", json!({"tool_name": "Write", "tool_input": {"file_path": "/w/app/d.rs"}})),
485 ];
486 for (i, (e, input)) in steps.iter().enumerate() {
487 p = Some(update(p, "corp", "claude", e, input, 1000 + i as i64, who));
488 }
489 let p = p.unwrap();
490 assert_eq!(p.prompt, "fix the flaky sync test", "first non-empty line");
491 assert_eq!(p.files, ["d.rs", "/elsewhere/c.rs", "src/a.rs"], "newest first, deduped, at most 3, relative inside cwd");
492 assert_eq!((p.started, p.ms, p.pid, p.engineer.as_str(), p.repo.as_str()), (1000, 1006, Some(4242), "alice@x", "app"));
493 let long = "x".repeat(300);
494 assert_eq!(first_line(&long).chars().count(), PROMPT_CHARS + 1);
495 }
496
497 #[test]
498 fn write_read_prune_and_state() {
499 let home = tmp("wrp");
500 let input = json!({"session_id": "s/../1", "cwd": "/w/app", "prompt": "hello"});
501 let p = update(None, "corp", "omp", "prompt", &input, 5_000, who);
502 write(&home, &p).unwrap();
503 assert_eq!(read(&home, "s/../1"), Some(p.clone()));
504 assert_eq!(path(&home, "s/../1"), dir(&home).join("s1.json"), "a session id never escapes the directory");
505 assert_eq!(local(&home, 5_000), vec![p.clone()]);
506 assert!(std::fs::read_dir(dir(&home)).unwrap().count() == 1, "no temp file left behind");
507
508 assert_eq!(state(&p, 5_000, |_| true), State::Working);
509 assert_eq!(state(&Presence { event: "stop".into(), ..p.clone() }, 5_000, |_| true), State::Waiting);
510 assert_eq!(state(&p, 5_000 + IDLE_MS + 1, |_| true), State::Idle);
511 assert_eq!(state(&Presence { event: "end".into(), ..p.clone() }, 5_000, |_| true), State::Ended);
512 assert_eq!(state(&Presence { pid: Some(7), ..p.clone() }, 5_000, |_| false), State::Ended, "the harness process is gone");
513
514 assert!(local(&home, 5_000 + PRUNE_MS + 1).is_empty());
516 let far = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_millis() as i64 + PRUNE_MS + 1000;
517 prune(&home, far);
518 assert!(read(&home, "s/../1").is_none());
519 let _ = std::fs::remove_dir_all(&home);
520 }
521
522 #[test]
523 fn remote_keeps_the_latest_per_member_and_session() {
524 let home = tmp("remote");
525 std::fs::create_dir_all(home.join("mesh")).unwrap();
526 let p = |s: &str, ms: i64| Presence {
527 session: s.into(),
528 harness: "claude".into(),
529 engineer: "bob@x".into(),
530 repo: "r".into(),
531 collective: "corp".into(),
532 prompt: format!("at {ms}"),
533 files: vec![],
534 event: "prompt".into(),
535 ms,
536 started: 1,
537 pid: None,
538 };
539 let line = |s: &str, ms: i64, key: &str| serde_json::to_string(&Remote { p: p(s, ms), key: key.into(), sig: "00".into() }).unwrap();
540 let text = [line("a", 10, "k1"), line("a", 30, "k1"), line("a", 20, "k1"), line("a", 5, "k2"), "garbage".into()].join("\n");
541 std::fs::write(remote_path(&home), text).unwrap();
542 let r = remote(&home, 40);
543 assert_eq!(r.iter().map(|r| (r.key.as_str(), r.p.ms)).collect::<Vec<_>>(), [("k1", 30), ("k2", 5)]);
544 assert!(remote(&home, 40 + PRUNE_MS).is_empty(), "a day old: gone");
545 let _ = std::fs::remove_dir_all(&home);
546 }
547
548 #[test]
551 fn a_resumed_session_is_not_shown_its_messages_again() {
552 let home = tmp("resume");
553 let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_millis() as i64;
554 let input = json!({"session_id": "s2"});
555 write(&home, &update(None, "corp", "claude", "prompt", &input, now, who)).unwrap();
556 let from = crate::msg::Sender { by: "s1".into(), name: "Two of Five".into(), harness: "claude".into(), node: "n1".into() };
557 crate::msg::send(&home, crate::msg::Message::new(crate::msg::TELL, &from, "s2", None, now, "the API is stable"), now).unwrap();
558 let caps = crate::harness::by_name("claude").expect("a harness").caps();
559 assert!(crate::msg::deliver(&home, caps, "prompt", &input, Some(now), now, None).is_some(), "shown");
560
561 let later = now + PRUNE_MS + 60_000;
562 prune(&home, later);
563 assert!(read(&home, "s2").is_none(), "the presence is pruned");
564 write(&home, &update(None, "corp", "claude", "prompt", &input, later, who)).unwrap();
565 assert_eq!(crate::msg::deliver(&home, caps, "prompt", &input, Some(later), later, None), None, "not shown again");
566 let _ = std::fs::remove_dir_all(&home);
567 }
568}