use std::path::{Path, PathBuf};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use crate::learn;
pub const IDLE_MS: i64 = 30 * 60 * 1000;
pub const PRUNE_MS: i64 = crate::clock::DAY_MS;
pub const FILES: usize = 3;
const PROMPT_CHARS: usize = 100;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Presence {
pub session: String,
pub harness: String,
#[serde(default)]
pub engineer: String,
#[serde(default)]
pub repo: String,
pub collective: String,
#[serde(default)]
pub prompt: String,
#[serde(default)]
pub files: Vec<String>,
pub event: String,
pub ms: i64,
pub started: i64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub pid: Option<u32>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum State {
Working,
Waiting,
Idle,
Ended,
}
impl State {
pub fn name(self) -> &'static str {
match self {
State::Working => "working",
State::Waiting => "waiting",
State::Idle => "idle",
State::Ended => "ended",
}
}
}
pub fn dir(home: &Path) -> PathBuf {
home.join("presence")
}
pub fn safe(session: &str) -> String {
session.chars().filter(|c| c.is_ascii_alphanumeric() || *c == '-').collect()
}
pub fn path(home: &Path, session: &str) -> PathBuf {
dir(home).join(format!("{}.json", safe(session)))
}
pub fn remote_path(home: &Path) -> PathBuf {
home.join("mesh").join("presence.jsonl")
}
pub fn read(home: &Path, session: &str) -> Option<Presence> {
serde_json::from_slice(&std::fs::read(path(home, session)).ok()?).ok()
}
fn file_of(input: &Value, cwd: &str) -> Option<String> {
let ti = input.get("tool_input")?;
let f = ["file_path", "notebook_path"].iter().find_map(|k| ti.get(*k).and_then(Value::as_str)).filter(|f| !f.trim().is_empty())?;
let rel = Path::new(f).strip_prefix(cwd).ok().map(|p| p.display().to_string()).filter(|p| !p.is_empty());
Some(rel.unwrap_or_else(|| f.to_string()))
}
pub fn first_line(prompt: &str) -> String {
let line = prompt.lines().map(str::trim).find(|l| !l.is_empty()).unwrap_or_default();
let mut out: String = line.chars().take(PROMPT_CHARS).collect();
if line.chars().count() > PROMPT_CHARS {
out.push('…');
}
out
}
pub fn update(
prev: Option<Presence>,
collective: &str,
harness: &str,
event: &str,
input: &Value,
now: i64,
who: impl FnOnce(&str) -> (String, String),
) -> Presence {
let cwd = input.get("cwd").and_then(Value::as_str).unwrap_or_default();
let session = input.get("session_id").and_then(Value::as_str).unwrap_or("unknown");
let mut p = prev.unwrap_or_else(|| {
let (engineer, repo) = who(cwd);
Presence {
session: session.into(),
harness: harness.into(),
engineer,
repo,
collective: collective.into(),
prompt: String::new(),
files: Vec::new(),
event: String::new(),
ms: now,
started: now,
pid: None,
}
});
p.event = event.into();
p.ms = now;
if let Some(pid) = input.get("pid").and_then(Value::as_u64).and_then(|v| u32::try_from(v).ok()) {
p.pid = Some(pid);
}
match event {
"prompt" => {
let line = first_line(input.get("prompt").and_then(Value::as_str).unwrap_or_default());
if !line.is_empty() {
p.prompt = line;
}
}
"post_tool" | "post_tool_failure" => {
if let Some(f) = file_of(input, cwd) {
p.files.retain(|x| *x != f);
p.files.insert(0, f);
p.files.truncate(FILES);
}
}
_ => {}
}
p
}
pub fn touch(home: &Path, collective: &str, harness: &str, event: &str, input: &Value, now: i64) -> Option<Presence> {
let session = input.get("session_id").and_then(Value::as_str).unwrap_or("unknown");
let p = update(read(home, session), collective, harness, event, input, now, |cwd| {
(learn::engineer(cwd), borg_trace::repo_key(cwd).rsplit('/').next().unwrap_or_default().to_string())
});
write(home, &p).ok()?;
if event == "prompt" {
prune(home, now);
}
Some(p)
}
pub fn write(home: &Path, p: &Presence) -> std::io::Result<()> {
let d = dir(home);
std::fs::create_dir_all(&d)?;
let target = path(home, &p.session);
let tmp = d.join(format!(".{}.{}.tmp", safe(&p.session), std::process::id()));
std::fs::write(&tmp, serde_json::to_vec(p).unwrap_or_default())?;
std::fs::rename(&tmp, &target).inspect_err(|_| {
let _ = std::fs::remove_file(&tmp);
})
}
pub fn prune(home: &Path, now: i64) {
let Ok(rd) = std::fs::read_dir(dir(home)) else { return };
for e in rd.flatten() {
let old = e
.metadata()
.and_then(|m| m.modified())
.ok()
.and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
.is_some_and(|d| now - d.as_millis() as i64 > PRUNE_MS);
if old {
let _ = std::fs::remove_file(e.path());
if let Some(session) = e.path().file_stem().and_then(|s| s.to_str()).filter(|s| !s.starts_with('.')) {
crate::listen::forget(home, session);
}
}
}
}
pub fn local(home: &Path, now: i64) -> Vec<Presence> {
let Ok(rd) = std::fs::read_dir(dir(home)) else { return Vec::new() };
let mut v: Vec<Presence> = rd
.flatten()
.filter(|e| e.path().extension().is_some_and(|x| x == "json"))
.filter_map(|e| serde_json::from_slice::<Presence>(&std::fs::read(e.path()).ok()?).ok())
.filter(|p| now - p.ms <= PRUNE_MS)
.collect();
v.sort_by(|a, b| (a.started, &a.session).cmp(&(b.started, &b.session)));
v
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Remote {
pub p: Presence,
pub key: String,
pub sig: String,
}
pub fn remote(home: &Path, now: i64) -> Vec<Remote> {
let text = std::fs::read_to_string(remote_path(home)).unwrap_or_default();
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();
v.sort_by(|a, b| (&a.key, &a.p.session, b.p.ms).cmp(&(&b.key, &b.p.session, a.p.ms)));
v.dedup_by(|a, b| a.key == b.key && a.p.session == b.p.session);
v
}
pub fn synced_path(home: &Path) -> PathBuf {
home.join("mesh").join("synced.json")
}
pub fn synced(home: &Path) -> std::collections::BTreeMap<String, i64> {
std::fs::read_to_string(synced_path(home)).ok().and_then(|t| serde_json::from_str(&t).ok()).unwrap_or_default()
}
pub fn note_synced(home: &Path, key: &str, ms: i64) -> std::io::Result<()> {
let mut all = synced(home);
if all.get(key).is_some_and(|&held| held >= ms && held <= crate::learn::now_ms()) {
return Ok(());
}
all.insert(key.into(), ms);
write_synced(home, &all)
}
pub fn forget_synced(home: &Path, key: &str) -> std::io::Result<()> {
let mut all = synced(home);
if all.remove(key).is_none() {
return Ok(());
}
write_synced(home, &all)
}
fn write_synced(home: &Path, all: &std::collections::BTreeMap<String, i64>) -> std::io::Result<()> {
write_mesh(home, &synced_path(home), all)
}
pub(crate) fn write_mesh<T: Serialize>(home: &Path, p: &Path, all: &T) -> std::io::Result<()> {
std::fs::create_dir_all(home.join("mesh"))?;
let tmp = p.with_extension(format!("{}.tmp", std::process::id())); std::fs::write(&tmp, serde_json::to_string(all).unwrap_or_default())?;
std::fs::rename(&tmp, p)
}
pub use crate::protocol::{Mismatch, forget_mismatch, mismatches, note_mismatch, protocols_path, retain_mismatches};
pub fn reads_path(home: &Path) -> PathBuf {
home.join("mesh").join("reads.json")
}
pub fn reads(home: &Path) -> std::collections::BTreeMap<String, u8> {
std::fs::read_to_string(reads_path(home)).ok().and_then(|t| serde_json::from_str(&t).ok()).unwrap_or_default()
}
pub fn note_reads(home: &Path, key: &str, v: u8) -> std::io::Result<()> {
let mut all = reads(home);
if all.get(key) == Some(&v) {
return Ok(());
}
all.insert(key.into(), v);
write_mesh(home, &reads_path(home), &all)
}
pub fn retain_reads(home: &Path, keep: impl Fn(&str) -> bool) -> std::io::Result<()> {
let mut all = reads(home);
let before = all.len();
all.retain(|k, _| keep(k));
if all.len() == before {
return Ok(());
}
write_mesh(home, &reads_path(home), &all)
}
pub fn mesh_checks(routes: &[crate::collective::Route]) -> Vec<(crate::harness::Level, String)> {
use crate::collective::Route;
routes
.iter()
.filter_map(|r| match r {
Route::Collective { name, home, .. } => Some((name, home)),
Route::Off => None,
})
.flat_map(|(name, home)| {
let mismatched = mismatches(home);
let versions: Vec<_> = crate::protocol::versions(home)
.into_iter()
.filter(|(k, _)| !mismatched.contains_key(k))
.filter_map(|(k, v)| crate::protocol::check(&k, &v, crate::update::VERSION))
.map(|(level, line)| (level, format!("mesh {name}: {line}")))
.collect();
let lines = mismatched.into_iter().map(move |(k, m)| (crate::harness::Level::Warn, format!("mesh {name}: {}", m.line(&k))));
lines.chain(versions)
})
.collect()
}
pub fn state(p: &Presence, now: i64, alive: impl Fn(u32) -> bool) -> State {
if p.event == "end" || p.pid.is_some_and(|pid| !alive(pid)) {
State::Ended
} else if now - p.ms > IDLE_MS {
State::Idle
} else if p.event == "stop" {
State::Waiting
} else {
State::Working
}
}
pub fn alive(pid: u32) -> bool {
i32::try_from(pid).is_ok_and(crate::live::alive)
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
fn tmp(tag: &str) -> PathBuf {
let d = std::env::temp_dir().join(format!("borg-presence-{tag}-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&d);
d
}
fn who(_: &str) -> (String, String) {
("alice@x".into(), "app".into())
}
#[test]
fn a_members_last_session_never_moves_back() {
let h = tmp("synced");
assert!(synced(&h).is_empty());
note_synced(&h, "k1", 200).unwrap();
note_synced(&h, "k1", 100).unwrap();
note_synced(&h, "k2", 50).unwrap();
assert_eq!(synced(&h).into_iter().collect::<Vec<_>>(), [("k1".to_string(), 200), ("k2".to_string(), 50)]);
note_synced(&h, "k1", 300).unwrap();
assert_eq!(synced(&h)["k1"], 300);
forget_synced(&h, "k1").unwrap();
assert_eq!(synced(&h).into_keys().collect::<Vec<_>>(), ["k2"], "a failed dial drops the member");
let _ = std::fs::remove_dir_all(&h);
}
#[test]
fn doctor_names_a_member_on_another_mesh_protocol() {
use crate::collective::Route;
use crate::harness::Level;
let home = tmp("protocols-doctor");
let route = Route::Collective { name: "team".into(), home: home.clone(), share: crate::collective::Share::Team };
assert!(mesh_checks(std::slice::from_ref(&route)).is_empty());
let m = Mismatch { theirs: "borg/mesh/5".into(), ours: "borg/mesh/10".into(), ms: 1 };
note_mismatch(&home, "abcdef0123456789", m).unwrap();
let checks = mesh_checks(&[route, Route::Off]);
assert_eq!(checks.len(), 1, "{checks:?}");
assert_eq!(checks[0].0, Level::Warn);
assert!(checks[0].1.starts_with("mesh team: peer abcdef012345 speaks borg/mesh/5") && checks[0].1.ends_with("upgrade it"), "{checks:?}");
let n = crate::protocol::current();
let v = |borg: &str, protocol| crate::protocol::PeerVersion { borg: Some(borg.into()), protocol, ms: 1, engineer: None };
crate::protocol::note_version(&home, "k-old", &v("0.6.0", n - 1)).unwrap();
crate::protocol::note_version(&home, "k-new", &v("99.0.0", n)).unwrap();
crate::protocol::note_version(&home, "abcdef0123456789", &v("0.1.0", n - 9)).unwrap();
let checks = mesh_checks(std::slice::from_ref(&Route::Collective { name: "team".into(), home: home.clone(), share: crate::collective::Share::Team }));
assert_eq!(checks.len(), 3, "a mismatched member is named once: {checks:?}");
assert_eq!(
(checks[1].0, checks[1].1.as_str()),
(Level::Ok, format!("mesh team: peer k-new runs borg 99.0.0 (borg/mesh/{n}), this node {}", crate::update::VERSION).as_str())
);
assert_eq!(checks[2].0, Level::Warn);
assert!(checks[2].1.ends_with(crate::protocol::note(n).unwrap()), "{checks:?}");
let _ = std::fs::remove_dir_all(&home);
}
#[test]
fn a_protocol_mismatch_is_kept_until_a_session_completes() {
let h = tmp("protocols");
assert!(mismatches(&h).is_empty());
let old = Mismatch { theirs: "borg/mesh/5".into(), ours: "borg/mesh/10".into(), ms: 1 };
note_mismatch(&h, "abcdef0123456789", old.clone()).unwrap();
note_mismatch(&h, "k2", Mismatch { theirs: "borg/mesh/11".into(), ..old.clone() }).unwrap();
let all = mismatches(&h);
assert_eq!(
all["abcdef0123456789"].line("abcdef0123456789"),
"peer abcdef012345 speaks borg/mesh/5, this node borg/mesh/10: nothing syncs with it; upgrade it"
);
assert!(all["k2"].line("k2").ends_with("upgrade this node"), "a newer peer: this node is the old one");
forget_mismatch(&h, "abcdef0123456789").unwrap();
assert_eq!(mismatches(&h).into_keys().collect::<Vec<_>>(), ["k2"]);
let _ = std::fs::remove_dir_all(&h);
}
#[test]
fn a_record_ahead_of_the_clock_gives_way() {
let h = tmp("synced-skew");
let now = crate::learn::now_ms();
note_synced(&h, "k1", now + 3_600_000).unwrap();
note_synced(&h, "k1", now).unwrap();
assert_eq!(synced(&h)["k1"], now);
let _ = std::fs::remove_dir_all(&h);
}
#[test]
fn a_session_record_follows_its_hooks() {
let ev = |e: &str, extra: Value| {
let mut v = json!({"session_id": "s1", "cwd": "/w/app"});
v.as_object_mut().unwrap().extend(extra.as_object().unwrap().clone());
(e.to_string(), v)
};
let mut p = None;
let steps = [
ev("prompt", json!({"prompt": "\n fix the flaky sync test\nand more", "pid": 4242})),
ev("post_tool", json!({"tool_name": "Edit", "tool_input": {"file_path": "/w/app/src/a.rs"}})),
ev("post_tool", json!({"tool_name": "Read", "tool_input": {"file_path": "/w/app/src/b.rs"}})),
ev("post_tool", json!({"tool_name": "Bash", "tool_input": {"command": "ls"}})),
ev("post_tool", json!({"tool_name": "Edit", "tool_input": {"file_path": "/w/app/src/a.rs"}})),
ev("post_tool", json!({"tool_name": "Write", "tool_input": {"file_path": "/elsewhere/c.rs"}})),
ev("post_tool", json!({"tool_name": "Write", "tool_input": {"file_path": "/w/app/d.rs"}})),
];
for (i, (e, input)) in steps.iter().enumerate() {
p = Some(update(p, "corp", "claude", e, input, 1000 + i as i64, who));
}
let p = p.unwrap();
assert_eq!(p.prompt, "fix the flaky sync test", "first non-empty line");
assert_eq!(p.files, ["d.rs", "/elsewhere/c.rs", "src/a.rs"], "newest first, deduped, at most 3, relative inside cwd");
assert_eq!((p.started, p.ms, p.pid, p.engineer.as_str(), p.repo.as_str()), (1000, 1006, Some(4242), "alice@x", "app"));
let long = "x".repeat(300);
assert_eq!(first_line(&long).chars().count(), PROMPT_CHARS + 1);
}
#[test]
fn write_read_prune_and_state() {
let home = tmp("wrp");
let input = json!({"session_id": "s/../1", "cwd": "/w/app", "prompt": "hello"});
let p = update(None, "corp", "omp", "prompt", &input, 5_000, who);
write(&home, &p).unwrap();
assert_eq!(read(&home, "s/../1"), Some(p.clone()));
assert_eq!(path(&home, "s/../1"), dir(&home).join("s1.json"), "a session id never escapes the directory");
assert_eq!(local(&home, 5_000), vec![p.clone()]);
assert!(std::fs::read_dir(dir(&home)).unwrap().count() == 1, "no temp file left behind");
assert_eq!(state(&p, 5_000, |_| true), State::Working);
assert_eq!(state(&Presence { event: "stop".into(), ..p.clone() }, 5_000, |_| true), State::Waiting);
assert_eq!(state(&p, 5_000 + IDLE_MS + 1, |_| true), State::Idle);
assert_eq!(state(&Presence { event: "end".into(), ..p.clone() }, 5_000, |_| true), State::Ended);
assert_eq!(state(&Presence { pid: Some(7), ..p.clone() }, 5_000, |_| false), State::Ended, "the harness process is gone");
assert!(local(&home, 5_000 + PRUNE_MS + 1).is_empty());
let far = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_millis() as i64 + PRUNE_MS + 1000;
prune(&home, far);
assert!(read(&home, "s/../1").is_none());
let _ = std::fs::remove_dir_all(&home);
}
#[test]
fn remote_keeps_the_latest_per_member_and_session() {
let home = tmp("remote");
std::fs::create_dir_all(home.join("mesh")).unwrap();
let p = |s: &str, ms: i64| Presence {
session: s.into(),
harness: "claude".into(),
engineer: "bob@x".into(),
repo: "r".into(),
collective: "corp".into(),
prompt: format!("at {ms}"),
files: vec![],
event: "prompt".into(),
ms,
started: 1,
pid: None,
};
let line = |s: &str, ms: i64, key: &str| serde_json::to_string(&Remote { p: p(s, ms), key: key.into(), sig: "00".into() }).unwrap();
let text = [line("a", 10, "k1"), line("a", 30, "k1"), line("a", 20, "k1"), line("a", 5, "k2"), "garbage".into()].join("\n");
std::fs::write(remote_path(&home), text).unwrap();
let r = remote(&home, 40);
assert_eq!(r.iter().map(|r| (r.key.as_str(), r.p.ms)).collect::<Vec<_>>(), [("k1", 30), ("k2", 5)]);
assert!(remote(&home, 40 + PRUNE_MS).is_empty(), "a day old: gone");
let _ = std::fs::remove_dir_all(&home);
}
#[test]
fn a_resumed_session_is_not_shown_its_messages_again() {
let home = tmp("resume");
let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_millis() as i64;
let input = json!({"session_id": "s2"});
write(&home, &update(None, "corp", "claude", "prompt", &input, now, who)).unwrap();
let from = crate::msg::Sender { by: "s1".into(), name: "Two of Five".into(), harness: "claude".into(), node: "n1".into() };
crate::msg::send(&home, crate::msg::Message::new(crate::msg::TELL, &from, "s2", None, now, "the API is stable"), now).unwrap();
let caps = crate::harness::by_name("claude").expect("a harness").caps();
assert!(crate::msg::deliver(&home, caps, "prompt", &input, Some(now), now, None).is_some(), "shown");
let later = now + PRUNE_MS + 60_000;
prune(&home, later);
assert!(read(&home, "s2").is_none(), "the presence is pruned");
write(&home, &update(None, "corp", "claude", "prompt", &input, later, who)).unwrap();
assert_eq!(crate::msg::deliver(&home, caps, "prompt", &input, Some(later), later, None), None, "not shown again");
let _ = std::fs::remove_dir_all(&home);
}
}