presence.rs 96.6%

171 of 177 lines run, 37 of 45 functions · run #17 on main · run missed

//! Presence and sessions (spec §8.1, §8.2), kept in memory by the daemon
//! from what agents do, and recorded as events when they change, so the
//! ledger says when an agent went quiet.

use crate::event::NewEvent;
use std::collections::HashMap;

/// A request within this long makes an agent online.
pub const ONLINE_MS: u64 = 60_000;
/// Silence this long ends a session and makes an agent offline.
pub const IDLE_AFTER_MS: u64 = 10 * 60_000;

#[derive(Clone, Debug, Default)]
struct Activity {
    last_request_ms: Option<u64>,
    waiting: u32,
    shims: u32,
    status_ms: Option<u64>,
    blocked: bool,
    paused: bool,
    session: Option<String>,
    state: Option<&'static str>,
}

/// Every agent's activity, and the state last recorded for each.
#[derive(Debug, Default)]
pub struct Presence {
    agents: HashMap<String, Activity>,
}

impl Presence {
    /// A request from `agent`: returns its session id, and a
    /// `agent.session_started` event when this begins a new one.
    pub fn request(&mut self, agent: &str, now: u64, new_session: bool) -> (String, Option<NewEvent>) {
        let a = self.agents.entry(agent.to_string()).or_default();
        let lapsed = a.last_request_ms.is_none_or(|t| now.saturating_sub(t) >= IDLE_AFTER_MS);
        a.last_request_ms = Some(now);
        if a.session.is_some() && !lapsed && !new_session {
            return (a.session.clone().unwrap_or_default(), None);
        }
        let id = crate::ids::ulid();
        a.session = Some(id.clone());
        let e = NewEvent::system("agent.session_started", serde_json::json!({ "agent": agent, "session": id }))
            .subject(agent)
            .session(id.clone());
        (id, Some(e))
    }
    pub fn wait_started(&mut self, agent: &str) {
        self.agents.entry(agent.to_string()).or_default().waiting += 1;
    }
    pub fn wait_ended(&mut self, agent: &str) {
        let a = self.agents.entry(agent.to_string()).or_default();
        a.waiting = a.waiting.saturating_sub(1);
    }
    pub fn shim_connected(&mut self, agent: &str) {
        self.agents.entry(agent.to_string()).or_default().shims += 1;
    }
    pub fn shim_gone(&mut self, agent: &str) {
        let a = self.agents.entry(agent.to_string()).or_default();
        a.shims = a.shims.saturating_sub(1);
    }
    /// The agent set its status (the `status` tool); `None` clears it.
    pub fn status(&mut self, agent: &str, at: Option<u64>) {
        self.agents.entry(agent.to_string()).or_default().status_ms = at;
    }
    pub fn blocked(&mut self, agent: &str, blocked: bool) {
        self.agents.entry(agent.to_string()).or_default().blocked = blocked;
    }
    pub fn paused(&mut self, agent: &str, paused: bool) {
        self.agents.entry(agent.to_string()).or_default().paused = paused;
    }

    /// The state `agent` is in at `now`.
    pub fn state(&self, agent: &str, now: u64) -> &'static str {
        self.agents.get(agent).map(|a| compute(a, now)).unwrap_or("offline")
    }

    /// Events for every change since the last call: `agent.presence` for a
    /// new state, `agent.session_ended` for a session gone quiet.
    pub fn tick(&mut self, now: u64) -> Vec<NewEvent> {
        let mut out = Vec::new();
        let mut names: Vec<&String> = self.agents.keys().collect();
        names.sort();
        let names: Vec<String> = names.into_iter().cloned().collect();
        for name in names {
            let a = self.agents.get_mut(&name).expect("listed");
            let s = compute(a, now);
            if a.state != Some(s) {
                let from = a.state.unwrap_or("offline");
                a.state = Some(s);
                if from != s {
                    out.push(
                        NewEvent::system("agent.presence", serde_json::json!({ "agent": name, "state": s, "from": from }))
                            .subject(name.clone()),
                    );
                }
            }
            let quiet = a.last_request_ms.is_none_or(|t| now.saturating_sub(t) >= IDLE_AFTER_MS);
            if quiet && a.waiting == 0 {
                if let Some(id) = a.session.take() {
                    out.push(
                        NewEvent::system("agent.session_ended", serde_json::json!({ "agent": name, "session": id }))
                            .subject(name.clone())
                            .session(id),
                    );
                }
            }
        }
        out
    }
}

/// Paused, then blocked, then busy, then online, then idle, then offline.
fn compute(a: &Activity, now: u64) -> &'static str {
    let within = |t: Option<u64>, ms: u64| t.is_some_and(|t| now.saturating_sub(t) < ms);
    if a.paused {
        "paused"
    } else if a.blocked {
        "blocked"
    } else if within(a.status_ms, IDLE_AFTER_MS) {
        "busy"
    } else if a.waiting > 0 || within(a.last_request_ms, ONLINE_MS) {
        "online"
    } else if a.shims > 0 || within(a.last_request_ms, IDLE_AFTER_MS) {
        "idle"
    } else {
        "offline"
    }
}

/// Apply presence, status and session events to their tables.
pub async fn project(conn: &mut sqlx::SqliteConnection, e: &crate::event::Event) -> sqlx::Result<()> {
    let p = e.json();
    let s = |k: &str| p.get(k).and_then(|v| v.as_str()).map(String::from);
    // A projection skips what it cannot use: a malformed payload must never
    // stop the ledger from taking events.
    match e.kind.as_str() {
        "agent.presence" if s("agent").is_some() && s("state").is_some() => {
            sqlx::query("INSERT INTO presence (agent, state, since_ms) VALUES (?, ?, ?) ON CONFLICT(agent) DO UPDATE SET state = excluded.state, since_ms = excluded.since_ms")
                .bind(s("agent"))
                .bind(s("state"))
                .bind(e.at_ms as i64)
                .execute(&mut *conn)
                .await?;
        }
        "agent.status" if e.actor.starts_with("agent:") => {
            let agent = e.actor.strip_prefix("agent:").map(String::from);
            sqlx::query(
                "INSERT INTO presence (agent, state, since_ms, status_state, status_text, status_task, status_ms) VALUES (?, 'online', ?, ?, ?, ?, ?) \
                 ON CONFLICT(agent) DO UPDATE SET status_state = excluded.status_state, status_text = excluded.status_text, status_task = excluded.status_task, status_ms = excluded.status_ms",
            )
            .bind(agent)
            .bind(e.at_ms as i64)
            .bind(s("state"))
            .bind(s("text"))
            .bind(s("task"))
            .bind(e.at_ms as i64)
            .execute(&mut *conn)
            .await?;
        }
        "agent.session_started" if s("agent").is_some() && s("session").is_some() => {
            sqlx::query("INSERT OR IGNORE INTO sessions (id, agent, started_ms) VALUES (?, ?, ?)")
                .bind(s("session"))
                .bind(s("agent"))
                .bind(e.at_ms as i64)
                .execute(&mut *conn)
                .await?;
        }
        "agent.session_ended" => {
            sqlx::query("UPDATE sessions SET ended_ms = ? WHERE id = ?")
                .bind(e.at_ms as i64)
                .bind(s("session"))
                .execute(&mut *conn)
                .await?;
        }
        _ => {}
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    fn kinds(v: &[NewEvent]) -> Vec<String> {
        v.iter().map(|e| format!("{} {}", e.kind, e.payload.get("state").and_then(|s| s.as_str()).unwrap_or(""))).collect()
    }

    #[test]
    fn presence_moves_through_its_states_and_records_each_change() {
        let mut p = Presence::default();
        let t = 1_000_000;
        let (s1, started) = p.request("claude", t, false);
        assert!(started.is_some());
        assert_eq!(kinds(&p.tick(t)), ["agent.presence online"]);
        assert!(p.tick(t + 1).is_empty(), "no change, no event");
        assert_eq!(p.state("claude", t + ONLINE_MS + 1), "idle");
        assert_eq!(kinds(&p.tick(t + ONLINE_MS + 1)), ["agent.presence idle"]);
        p.status("claude", Some(t + ONLINE_MS + 2));
        assert_eq!(p.state("claude", t + ONLINE_MS + 3), "busy");
        p.blocked("claude", true);
        assert_eq!(p.state("claude", t + ONLINE_MS + 3), "blocked");
        p.blocked("claude", false);
        p.status("claude", None);
        p.paused("claude", true);
        assert_eq!(p.state("claude", t + ONLINE_MS + 3), "paused");
        p.paused("claude", false);
        let late = t + IDLE_AFTER_MS + 5;
        let ev = p.tick(late);
        assert_eq!(kinds(&ev), ["agent.presence offline", "agent.session_ended "]);
        let (s2, started) = p.request("claude", late + 1, false);
        assert!(started.is_some() && s2 != s1, "a new session after ten minutes' silence");
        let (s3, none) = p.request("claude", late + 2, false);
        assert!(none.is_none() && s3 == s2);
        let (s4, again) = p.request("claude", late + 3, true);
        assert!(again.is_some() && s4 != s3, "brief(new_session: true)");
    }

    #[test]
    fn a_waiting_or_connected_agent_is_not_offline() {
        let mut p = Presence::default();
        p.request("kimi", 0, false);
        p.wait_started("kimi");
        assert_eq!(p.state("kimi", IDLE_AFTER_MS * 3), "online");
        p.wait_ended("kimi");
        p.shim_connected("kimi");
        assert_eq!(p.state("kimi", IDLE_AFTER_MS * 3), "idle");
        p.shim_gone("kimi");
        assert_eq!(p.state("kimi", IDLE_AFTER_MS * 3), "offline");
    }
}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

1		//! Presence and sessions (spec §8.1, §8.2), kept in memory by the daemon
2		//! from what agents do, and recorded as events when they change, so the
3		//! ledger says when an agent went quiet.
4
5		use crate::event::NewEvent;
6		use std::collections::HashMap;
7
8		/// A request within this long makes an agent online.
9		pub const ONLINE_MS: u64 = 60_000;
10		/// Silence this long ends a session and makes an agent offline.
11		pub const IDLE_AFTER_MS: u64 = 10 * 60_000;
12
13		#[derive(Clone, Debug, Default)]
14		struct Activity {
15		last_request_ms: Option<u64>,
16		waiting: u32,
17		shims: u32,
18		status_ms: Option<u64>,
19		blocked: bool,
20		paused: bool,
21		session: Option<String>,
22		state: Option<&'static str>,
23		}
24
25		/// Every agent's activity, and the state last recorded for each.
26		#[derive(Debug, Default)]
27		pub struct Presence {
28		agents: HashMap<String, Activity>,
29		}
30
31		impl Presence {
32		/// A request from `agent`: returns its session id, and a
33		/// `agent.session_started` event when this begins a new one.
34	45×	pub fn request(&mut self, agent: &str, now: u64, new_session: bool) -> (String, Option<NewEvent>) {
35	45×	let a = self.agents.entry(agent.to_string()).or_default();
36	45×	let lapsed = a.last_request_ms.is_none_or(\|t\| now.saturating_sub(t) >= IDLE_AFTER_MS);
37	45×	a.last_request_ms = Some(now);
38	45×	if a.session.is_some() && !lapsed && !new_session {
39	30×	return (a.session.clone().unwrap_or_default(), None);
40	15×	}
41	15×	let id = crate::ids::ulid();
42	15×	a.session = Some(id.clone());
43	15×	let e = NewEvent::system("agent.session_started", serde_json::json!({ "agent": agent, "session": id }))
44	15×	.subject(agent)
45	15×	.session(id.clone());
46	15×	(id, Some(e))
47	45×	}
48	6×	pub fn wait_started(&mut self, agent: &str) {
49	6×	self.agents.entry(agent.to_string()).or_default().waiting += 1;
50	6×	}
51	6×	pub fn wait_ended(&mut self, agent: &str) {
52	6×	let a = self.agents.entry(agent.to_string()).or_default();
53	6×	a.waiting = a.waiting.saturating_sub(1);
54	6×	}
55	1×	pub fn shim_connected(&mut self, agent: &str) {
56	1×	self.agents.entry(agent.to_string()).or_default().shims += 1;
57	1×	}
58	1×	pub fn shim_gone(&mut self, agent: &str) {
59	1×	let a = self.agents.entry(agent.to_string()).or_default();
60	1×	a.shims = a.shims.saturating_sub(1);
61	1×	}
62		/// The agent set its status (the `status` tool); `None` clears it.
63	3×	pub fn status(&mut self, agent: &str, at: Option<u64>) {
64	3×	self.agents.entry(agent.to_string()).or_default().status_ms = at;
65	3×	}
66	3×	pub fn blocked(&mut self, agent: &str, blocked: bool) {
67	3×	self.agents.entry(agent.to_string()).or_default().blocked = blocked;
68	3×	}
69	2×	pub fn paused(&mut self, agent: &str, paused: bool) {
70	2×	self.agents.entry(agent.to_string()).or_default().paused = paused;
71	2×	}
72
73		/// The state `agent` is in at `now`.
74	20×	pub fn state(&self, agent: &str, now: u64) -> &'static str {
75	20×	self.agents.get(agent).map(\|a\| compute(a, now)).unwrap_or("offline")
76	20×	}
77
78		/// Events for every change since the last call: `agent.presence` for a
79		/// new state, `agent.session_ended` for a session gone quiet.
80	14×	pub fn tick(&mut self, now: u64) -> Vec<NewEvent> {
81	14×	let mut out = Vec::new();
82	14×	let mut names: Vec<&String> = self.agents.keys().collect();
83	14×	names.sort();
84	14×	let names: Vec<String> = names.into_iter().cloned().collect();
85	14×	for name in names {
86	4×	let a = self.agents.get_mut(&name).expect("listed");
87	4×	let s = compute(a, now);
88	4×	if a.state != Some(s) {
89	3×	let from = a.state.unwrap_or("offline");
90	3×	a.state = Some(s);
91	3×	if from != s {
92	3×	out.push(
93	3×	NewEvent::system("agent.presence", serde_json::json!({ "agent": name, "state": s, "from": from }))
94	3×	.subject(name.clone()),
95	3×	);
96	3×	}
97	1×	}
98	4×	let quiet = a.last_request_ms.is_none_or(\|t\| now.saturating_sub(t) >= IDLE_AFTER_MS);
99	4×	if quiet && a.waiting == 0 {
100	1×	if let Some(id) = a.session.take() {
101	1×	out.push(
102	1×	NewEvent::system("agent.session_ended", serde_json::json!({ "agent": name, "session": id }))
103	1×	.subject(name.clone())
104	1×	.session(id),
105	1×	);
106	1×	}
107	3×	}
108		}
109	14×	out
110	14×	}
111		}
112
113		/// Paused, then blocked, then busy, then online, then idle, then offline.
114	19×	fn compute(a: &Activity, now: u64) -> &'static str {
115	35×	let within = \|t: Option<u64>, ms: u64\| t.is_some_and(\|t\| now.saturating_sub(t) < ms);
116	19×	if a.paused {
117	1×	"paused"
118	18×	} else if a.blocked {
119	1×	"blocked"
120	17×	} else if within(a.status_ms, IDLE_AFTER_MS) {
121	2×	"busy"
122	15×	} else if a.waiting > 0 \|\| within(a.last_request_ms, ONLINE_MS) {
123	10×	"online"
124	5×	} else if a.shims > 0 \|\| within(a.last_request_ms, IDLE_AFTER_MS) {
125	3×	"idle"
126		} else {
127	2×	"offline"
128		}
129	19×	}
130
131		/// Apply presence, status and session events to their tables.
132	42500×	pub async fn project(conn: &mut sqlx::SqliteConnection, e: &crate::event::Event) -> sqlx::Result<()> {
133	42500×	let p = e.json();
134	42500×	let s = \|k: &str\| p.get(k).and_then(\|v\| v.as_str()).map(String::from);
135		// A projection skips what it cannot use: a malformed payload must never
136		// stop the ledger from taking events.
137	42500×	match e.kind.as_str() {
138	42500×	"agent.presence" if s("agent").is_some() && s("state").is_some() => {
139	0×	sqlx::query("INSERT INTO presence (agent, state, since_ms) VALUES (?, ?, ?) ON CONFLICT(agent) DO UPDATE SET state = excluded.state, since_ms = excluded.since_ms")
140	0×	.bind(s("agent"))
141	0×	.bind(s("state"))
142	0×	.bind(e.at_ms as i64)
143	0×	.execute(&mut *conn)
144	0×	.await?;
145		}
146	38988×	"agent.status" if e.actor.starts_with("agent:") => {
147	1×	let agent = e.actor.strip_prefix("agent:").map(String::from);
148	1×	sqlx::query(
149	1×	"INSERT INTO presence (agent, state, since_ms, status_state, status_text, status_task, status_ms) VALUES (?, 'online', ?, ?, ?, ?, ?) \
150	1×	ON CONFLICT(agent) DO UPDATE SET status_state = excluded.status_state, status_text = excluded.status_text, status_task = excluded.status_task, status_ms = excluded.status_ms",
151	1×	)
152	1×	.bind(agent)
153	1×	.bind(e.at_ms as i64)
154	1×	.bind(s("state"))
155	1×	.bind(s("text"))
156	1×	.bind(s("task"))
157	1×	.bind(e.at_ms as i64)
158	1×	.execute(&mut *conn)
159	1×	.await?;
160		}
161	35595×	"agent.session_started" if s("agent").is_some() && s("session").is_some() => {
162	11×	sqlx::query("INSERT OR IGNORE INTO sessions (id, agent, started_ms) VALUES (?, ?, ?)")
163	11×	.bind(s("session"))
164	11×	.bind(s("agent"))
165	11×	.bind(e.at_ms as i64)
166	11×	.execute(&mut *conn)
167	11×	.await?;
168		}
169	32328×	"agent.session_ended" => {
170	3476×	sqlx::query("UPDATE sessions SET ended_ms = ? WHERE id = ?")
171	3476×	.bind(e.at_ms as i64)
172	3476×	.bind(s("session"))
173	3476×	.execute(&mut *conn)
174	3476×	.await?;
175		}
176	39012×	_ => {}
177		}
178	42500×	Ok(())
179	42500×	}
180
181		#[cfg(test)]
182		mod tests {
183		use super::*;
184
185	3×	fn kinds(v: &[NewEvent]) -> Vec<String> {
186	4×	v.iter().map(\|e\| format!("{} {}", e.kind, e.payload.get("state").and_then(\|s\| s.as_str()).unwrap_or(""))).collect()
187	3×	}
188
189		#[test]
190	1×	fn presence_moves_through_its_states_and_records_each_change() {
191	1×	let mut p = Presence::default();
192	1×	let t = 1_000_000;
193	1×	let (s1, started) = p.request("claude", t, false);
194	1×	assert!(started.is_some());
195	1×	assert_eq!(kinds(&p.tick(t)), ["agent.presence online"]);
196	1×	assert!(p.tick(t + 1).is_empty(), "no change, no event");
197	1×	assert_eq!(p.state("claude", t + ONLINE_MS + 1), "idle");
198	1×	assert_eq!(kinds(&p.tick(t + ONLINE_MS + 1)), ["agent.presence idle"]);
199	1×	p.status("claude", Some(t + ONLINE_MS + 2));
200	1×	assert_eq!(p.state("claude", t + ONLINE_MS + 3), "busy");
201	1×	p.blocked("claude", true);
202	1×	assert_eq!(p.state("claude", t + ONLINE_MS + 3), "blocked");
203	1×	p.blocked("claude", false);
204	1×	p.status("claude", None);
205	1×	p.paused("claude", true);
206	1×	assert_eq!(p.state("claude", t + ONLINE_MS + 3), "paused");
207	1×	p.paused("claude", false);
208	1×	let late = t + IDLE_AFTER_MS + 5;
209	1×	let ev = p.tick(late);
210	1×	assert_eq!(kinds(&ev), ["agent.presence offline", "agent.session_ended "]);
211	1×	let (s2, started) = p.request("claude", late + 1, false);
212	1×	assert!(started.is_some() && s2 != s1, "a new session after ten minutes' silence");
213	1×	let (s3, none) = p.request("claude", late + 2, false);
214	1×	assert!(none.is_none() && s3 == s2);
215	1×	let (s4, again) = p.request("claude", late + 3, true);
216	1×	assert!(again.is_some() && s4 != s3, "brief(new_session: true)");
217	1×	}
218
219		#[test]
220	1×	fn a_waiting_or_connected_agent_is_not_offline() {
221	1×	let mut p = Presence::default();
222	1×	p.request("kimi", 0, false);
223	1×	p.wait_started("kimi");
224	1×	assert_eq!(p.state("kimi", IDLE_AFTER_MS * 3), "online");
225	1×	p.wait_ended("kimi");
226	1×	p.shim_connected("kimi");
227	1×	assert_eq!(p.state("kimi", IDLE_AFTER_MS * 3), "idle");
228	1×	p.shim_gone("kimi");
229	1×	assert_eq!(p.state("kimi", IDLE_AFTER_MS * 3), "offline");
230	1×	}
231		}

Coverage / crates/spectra-core/src/presence.rs 96.6%

Keyboard