Skip to main content

borg/
presence.rs

1//! Who is working where, right now: one tiny record per drone session, for `borg top` (top.rs) and
2//! for steering (steer.rs) to name a drone.
3//!
4//! `<store>/presence/<session>.json`, rewritten whole by every hook call (one small write to a temp
5//! file, then a rename: a reader never sees half a record). No extra process, no network: the
6//! engineer (git user.email) and the repo are looked up once, on the session's first hook, and then
7//! carried over from the previous record.
8//!
9//! A session is `working` until its turn ends (`stop`: `waiting`), `idle` after 30 minutes without a
10//! hook, `ended` at `end` (or when the harness gave a pid that is gone). Records untouched for a day
11//! are pruned (at a prompt and by `borg top`). Remote drones (other members' sessions, carried by the
12//! mesh daemon when the collective may sync) are in `<store>/mesh/presence.jsonl`, never in lessons.
13
14use std::path::{Path, PathBuf};
15
16use serde::{Deserialize, Serialize};
17use serde_json::Value;
18
19use crate::learn;
20
21/// Without a hook for this long a session shows as idle.
22pub const IDLE_MS: i64 = 30 * 60 * 1000;
23/// Untouched for this long a record is pruned (and the mesh stops carrying it).
24pub const PRUNE_MS: i64 = 24 * 60 * 60 * 1000;
25/// At most this many recently touched files per record.
26pub const FILES: usize = 3;
27/// The prompt's first line, at most this many characters.
28const 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    /// git user.email in the session's repo, empty when unknown.
35    #[serde(default)]
36    pub engineer: String,
37    /// The repo's name (its key's last component).
38    #[serde(default)]
39    pub repo: String,
40    pub collective: String,
41    /// The first line of the engineer's latest prompt.
42    #[serde(default)]
43    pub prompt: String,
44    /// The files the drone touched last, newest first, relative to the session's directory.
45    #[serde(default)]
46    pub files: Vec<String>,
47    /// The canonical hook event last seen, and when (epoch ms).
48    pub event: String,
49    pub ms: i64,
50    /// When this record was first written: the drone's start, for its designation.
51    pub started: i64,
52    /// The harness process, if the harness gave it (`pid` in the hook event).
53    #[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
81/// Session ids become file names: only `[A-Za-z0-9-]` survives (as everywhere in the hook's state).
82pub 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
90/// Remote drones as the mesh daemon last heard of them (one signed record per line).
91pub 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
99/// What a tool call names as a file, relative to `cwd` when inside it.
100fn 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
107/// The prompt's first non-empty line, cut.
108pub 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
117/// The record after `event` (pure: `prev` is the session's last record; `who` looks up the engineer
118/// and repo, called only when there is no previous record).
119pub 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
170/// The hook's one call: updates the session's record with one atomic write. Returns the record
171/// (steering needs the session's start); `None` if it could not be written (fail open).
172pub 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
195/// Removes records not touched for a day (by file time: no parsing in the hook path).
196pub 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            // Its message claims stay: a resumed session keeps its id, and they must outlive the messages
207            // they cover (msg.rs drops them past the TTL).
208            let _ = std::fs::remove_file(e.path());
209            // Its opt-in to wakes (listen.rs) goes with it (the file stem is the safe session id).
210            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
217/// Every local record not yet due for pruning.
218pub 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/// A remote record as the mesh daemon stored it: the drone's presence and the member key that
231/// signed it (the daemon verified it on receipt).
232#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
233pub struct Remote {
234    pub p: Presence,
235    pub key: String,
236    pub sig: String,
237}
238
239/// The remote records not yet due for pruning (latest per member and session).
240pub 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
248/// `<store>/mesh/synced.json`, `{"<member key>": ms}`: when the last session with each member that
249/// completed started (this node's clock), written by the mesh daemon (borg-mesh sync.rs), dialled or
250/// served. Every line this node held before that ms was offered to the member in it, and the member's
251/// lines up to its own reading of the logs came back. Local, never synced.
252pub fn synced_path(home: &Path) -> PathBuf {
253    home.join("mesh").join("synced.json")
254}
255
256/// The members' last completed sessions (`synced_path`); empty when none completed yet.
257pub 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
261/// Records a completed session with `key` that started at `ms` (an older one never moves it back, but a
262/// record later than the clock now goes: the clock stepped back since, and it would stay ahead of
263/// every claim for as long). One small write to a temp file, then a rename: a reader never sees half
264/// of it. `serve` and a one-shot `borg mesh sync` take no lock between them: one may drop the other's
265/// newer entry, which only makes a claim wait for the next session.
266pub 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
275/// Drops `key`'s record (its last dial failed): no claim waits on it until a session completes again.
276pub 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
288/// One small write to a temp file, then a rename: a reader never sees half of `p`.
289fn 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())); // a one-shot `borg mesh sync` beside `serve`
292    std::fs::write(&tmp, serde_json::to_string(all).unwrap_or_default())?;
293    std::fs::rename(&tmp, p)
294}
295
296/// `<store>/mesh/protocols.json`, `{"<member key>": Mismatch}`: the members whose last session ended at
297/// the handshake because they speak another session protocol (borg-mesh sync.rs `ALPN`), so nothing
298/// syncs with them. Written by the mesh daemon; `borg doctor` names each one. A session with the member
299/// that completes drops its record, and so does a membership that no longer lists it. Local, never synced.
300pub fn protocols_path(home: &Path) -> PathBuf {
301    home.join("mesh").join("protocols.json")
302}
303
304/// A member's protocol beside this node's, both as their ALPN (`borg/mesh/<n>`), and when it was seen.
305#[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    /// What the mesh log and `borg doctor` say about member `key`: who has to upgrade.
314    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    /// The same, as the member on the other end reads it about this node (`me`): the close reason sent to it.
321    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
326/// The members whose protocol differs (`protocols_path`); empty when none.
327pub 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
331/// Records that `key` speaks another protocol (the newest sighting stays).
332pub 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
338/// Drops `key`'s record: a session with it completed, so it speaks this node's protocol now.
339pub 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
347/// Keeps only the records of keys `keep` still allows (the mesh calls it when the membership it holds
348/// changes: a removed or revoked member never completes another session, so nothing else drops it).
349pub 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
359/// `borg doctor`'s line for each member the mesh cannot sync with because it speaks another session
360/// protocol (`mismatches`, recorded by `borg mesh serve` or `sync`), in the stores of `routes`
361/// (install.rs `collectives`): a handshake that fails otherwise says nothing.
362pub 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
374/// The session's state at `now`; `alive` says whether a pid the harness gave still runs.
375pub 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
387/// Whether a local process runs (signal 0, so Linux and macOS alike).
388pub 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    /// Two sessions with one member can complete out of order (a served one beside a dialled one): the
408    /// later start stays.
409    #[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    /// A member on another protocol is recorded with who has to upgrade, until a session completes.
440    #[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        // The clock stepped back an hour after a session: the next session's start replaces it.
461        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        // A day later the record is no longer listed, and pruning (by file time) removes it.
515        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    /// A session resumed a day later keeps its id: the prune of its presence does not show its direct
549    /// messages again.
550    #[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}