crates/signald/src/lib.rs
487 lines · 20247 bytes
1//! # signald (library)
2//!
3//! The collector + transport internals of the ambient-companions daemon, split
4//! into a library so the collectors are unit-testable and the thin `signald`
5//! binary (`src/main.rs`) just wires arguments to the runtime.
6//!
7//! v0.2 implements the **git path** and the **terminal path** end to end, adds a
8//! **SQLite (WAL) history store** ([`history`]), and turns the socket into a
9//! **live pub/sub bus with a last-value cache** ([`hub`], [`publish`]). The
10//! sensitive terminal collector is aggregate-only and ships behind the now-active
11//! differential secret-typing test (see `signal-schema/tests/privacy_invariant.rs`).
12//! v0.4 adds the **hardware path**: the daemon spawns the sibling
13//! `macos-collector` and ingests its wire frames ([`collectors::hardware`]).
14
15use std::time::{SystemTime, UNIX_EPOCH};
16
17pub mod history;
18pub mod hub;
19
20/// Collectors: each reduces its domain to schema scalars (spec §1.1, §1.4).
21pub mod collectors {
22 /// Git collector: shell out to `git` for counts and ages. Counts and
23 /// branch/age scalars only — never diff content. This is the cleanest
24 /// signal in the suite and the first collector wired (spec §2.2, §3).
25 pub mod git {
26 use std::path::Path;
27 use std::process::Command;
28
29 use signal_schema::{Signal, SignalName, Source, Tag, Value, SCHEMA_VERSION};
30
31 use super::super::now_millis;
32
33 /// Rolling window (days) for [`SignalName::CommitsWindow`].
34 pub const WINDOW_DAYS: u64 = 7;
35
36 /// Derive the aggregate git signals for one repo. Every value is a
37 /// scalar `f64`; the only string on the wire is the repo path carried
38 /// in the audited `tag` (spec §1.2). Returns an empty vec if `repo` is
39 /// not a git repo.
40 pub fn collect(repo: &Path) -> Vec<Signal> {
41 if !is_git_repo(repo) {
42 return Vec::new();
43 }
44 let ts = now_millis();
45 let now_secs = ts / 1000;
46 let tag = Tag::repo_path(&repo.to_string_lossy());
47
48 let mk = |name: SignalName, value: f64| Signal {
49 schema_version: SCHEMA_VERSION,
50 ts,
51 source: Source::Git,
52 name,
53 value: Value(value),
54 tag: tag.clone(),
55 };
56
57 vec![
58 mk(
59 SignalName::CommitsWindow,
60 commits_since(repo, &format!("{WINDOW_DAYS} days ago")),
61 ),
62 mk(SignalName::CommitsToday, commits_since(repo, "midnight")),
63 mk(SignalName::BranchCount, branch_count(repo)),
64 mk(
65 SignalName::DaysSinceLastCommit,
66 days_since_last_commit(repo, now_secs),
67 ),
68 ]
69 }
70
71 fn is_git_repo(repo: &Path) -> bool {
72 run_git(repo, &["rev-parse", "--is-inside-work-tree"])
73 .map(|s| s == "true")
74 .unwrap_or(false)
75 }
76
77 /// Count commits reachable from `HEAD` newer than `since` (a git
78 /// approxidate, e.g. "midnight" or "7 days ago").
79 fn commits_since(repo: &Path, since: &str) -> f64 {
80 run_git(
81 repo,
82 &["rev-list", "--count", &format!("--since={since}"), "HEAD"],
83 )
84 .and_then(|s| s.parse::<f64>().ok())
85 .unwrap_or(0.0)
86 }
87
88 fn branch_count(repo: &Path) -> f64 {
89 run_git(repo, &["for-each-ref", "--format=%(refname)", "refs/heads/"])
90 .map(|s| if s.is_empty() { 0 } else { s.lines().count() })
91 .unwrap_or(0) as f64
92 }
93
94 /// Whole days between the most recent `HEAD` commit and now. `0.0` when
95 /// the repo has no commits yet.
96 fn days_since_last_commit(repo: &Path, now_secs: u64) -> f64 {
97 match run_git(repo, &["log", "-1", "--format=%ct", "HEAD"])
98 .and_then(|s| s.parse::<u64>().ok())
99 {
100 Some(last) => (now_secs.saturating_sub(last) / 86_400) as f64,
101 None => 0.0,
102 }
103 }
104
105 /// Run `git -C <repo> <args>` and return trimmed stdout on success.
106 fn run_git(repo: &Path, args: &[&str]) -> Option<String> {
107 let out = Command::new("git")
108 .arg("-C")
109 .arg(repo)
110 .args(args)
111 .output()
112 .ok()?;
113 if !out.status.success() {
114 return None;
115 }
116 Some(String::from_utf8_lossy(&out.stdout).trim().to_string())
117 }
118 }
119
120 /// Terminal collector (the sensitive one): **no input tap, no PTY sniff.**
121 ///
122 /// The shell side (`shell-hooks/signald-hooks.zsh`) counts keystrokes with a
123 /// `zle` widget that increments a *number* and discards the key, and appends
124 /// aggregate count records — `<epoch_ms> <keys> <session_seconds> <session_id>`,
125 /// numbers only — to a spool file on each `precmd`. Every shell appends to
126 /// the same spool; the session id (the shell's pid) keeps them apart.
127 ///
128 /// [`Collector`](terminal::Collector) consumes the spool each tick (renames
129 /// it aside, parses it, deletes it, so it never regrows) and keeps the last
130 /// two flushes of each session in memory. It derives
131 /// [`SignalName::KeysPerMin`](signal_schema::SignalName::KeysPerMin) as the
132 /// sum of each active session's rate and
133 /// [`SignalName::SessionSeconds`](signal_schema::SignalName::SessionSeconds)
134 /// as the longest active session. A line that is not four integers is
135 /// dropped, so the spool can carry nothing but counts (the privacy contract,
136 /// enforced by the differential secret-typing test).
137 pub mod terminal {
138 use std::collections::BTreeMap;
139 use std::path::{Path, PathBuf};
140
141 use signal_schema::{Signal, SignalName, Source, Value, SCHEMA_VERSION};
142
143 use super::super::now_millis;
144
145 /// A session with no flush inside this window is dropped from the
146 /// aggregates (its shell is idle or gone).
147 pub const ACTIVE_WINDOW_MS: u64 = 5 * 60_000;
148
149 /// One aggregate flush parsed from the spool: a timestamp, counts, and
150 /// the writing shell's session id. Every field is a number — there is
151 /// nowhere to put content.
152 #[derive(Debug, Clone, Copy, PartialEq)]
153 pub struct Flush {
154 pub ts_ms: u64,
155 pub keys: u64,
156 pub session_s: u64,
157 pub session: u64,
158 }
159
160 /// Parse one spool line. Returns `None` for anything that is not exactly
161 /// four integers, so non-count lines can never survive into a signal.
162 pub fn parse_flush(line: &str) -> Option<Flush> {
163 let mut it = line.split_whitespace();
164 let ts_ms = it.next()?.parse().ok()?;
165 let keys = it.next()?.parse().ok()?;
166 let session_s = it.next()?.parse().ok()?;
167 let session = it.next()?.parse().ok()?;
168 if it.next().is_some() {
169 return None; // extra fields => not a well-formed count record
170 }
171 Some(Flush {
172 ts_ms,
173 keys,
174 session_s,
175 session,
176 })
177 }
178
179 /// The previous and latest flush of one session.
180 #[derive(Debug, Clone, Copy)]
181 struct Session {
182 prev: Option<Flush>,
183 last: Flush,
184 }
185
186 impl Session {
187 /// Keys/min over the latest flush interval; the raw count when only
188 /// one flush (or a zero interval) is available.
189 fn keys_per_min(&self) -> f64 {
190 if let Some(prev) = self.prev {
191 let dt_ms = self.last.ts_ms.saturating_sub(prev.ts_ms);
192 if dt_ms > 0 {
193 return self.last.keys as f64 * 60_000.0 / dt_ms as f64;
194 }
195 }
196 self.last.keys as f64
197 }
198 }
199
200 /// Per-session state carried across ticks, since the spool is consumed.
201 #[derive(Debug, Default)]
202 pub struct Collector {
203 sessions: BTreeMap<u64, Session>,
204 }
205
206 impl Collector {
207 pub fn new() -> Collector {
208 Collector::default()
209 }
210
211 /// What the daemon does each tick: consume the spool, then derive
212 /// the current aggregates. Empty if no session is active.
213 pub fn collect(&mut self, spool: &Path) -> Vec<Signal> {
214 self.ingest(spool);
215 self.signals(now_millis())
216 }
217
218 /// Consume the spool: rename it aside (the hook opens it with `>>`
219 /// per write, so later appends land in a fresh file), parse, delete.
220 /// A leftover `.reading` file from an interrupted tick is read
221 /// first. Returns the number of records absorbed.
222 pub fn ingest(&mut self, spool: &Path) -> usize {
223 let reading = reading_path(spool);
224 if !reading.exists() && std::fs::rename(spool, &reading).is_err() {
225 return 0; // no spool yet
226 }
227 let text = std::fs::read_to_string(&reading).unwrap_or_default();
228 let _ = std::fs::remove_file(&reading);
229 self.absorb(text.lines().filter_map(parse_flush))
230 }
231
232 /// Fold flushes into per-session state (pure, unit-testable).
233 pub fn absorb(&mut self, flushes: impl IntoIterator<Item = Flush>) -> usize {
234 let mut n = 0;
235 for f in flushes {
236 n += 1;
237 self.sessions
238 .entry(f.session)
239 .and_modify(|s| {
240 s.prev = Some(s.last);
241 s.last = f;
242 })
243 .or_insert(Session { prev: None, last: f });
244 }
245 n
246 }
247
248 /// The aggregates at `now_ms`, dropping sessions idle longer than
249 /// [`ACTIVE_WINDOW_MS`]. Empty if no session is active.
250 pub fn signals(&mut self, now_ms: u64) -> Vec<Signal> {
251 self.sessions
252 .retain(|_, s| now_ms.saturating_sub(s.last.ts_ms) <= ACTIVE_WINDOW_MS);
253 if self.sessions.is_empty() {
254 return Vec::new();
255 }
256 let keys_per_min: f64 = self.sessions.values().map(Session::keys_per_min).sum();
257 let session_s = self.sessions.values().map(|s| s.last.session_s).max().unwrap_or(0);
258 let mk = |name: SignalName, value: f64| Signal {
259 schema_version: SCHEMA_VERSION,
260 ts: now_ms,
261 source: Source::Terminal,
262 name,
263 value: Value(value),
264 tag: None, // terminal aggregates are never tagged (spec §1.2)
265 };
266 vec![
267 mk(SignalName::KeysPerMin, keys_per_min),
268 mk(SignalName::SessionSeconds, session_s as f64),
269 ]
270 }
271 }
272
273 fn reading_path(spool: &Path) -> PathBuf {
274 let mut p = spool.as_os_str().to_os_string();
275 p.push(".reading");
276 PathBuf::from(p)
277 }
278
279 #[cfg(test)]
280 mod tests {
281 use super::*;
282
283 fn flush(ts_ms: u64, keys: u64, session_s: u64, session: u64) -> Flush {
284 Flush {
285 ts_ms,
286 keys,
287 session_s,
288 session,
289 }
290 }
291
292 #[test]
293 fn parse_requires_exactly_four_integers() {
294 assert_eq!(parse_flush("1000 12 30 4242"), Some(flush(1000, 12, 30, 4242)));
295 assert_eq!(parse_flush("1000 12 30"), None);
296 assert_eq!(parse_flush("1000 12 30 4242 extra"), None);
297 assert_eq!(parse_flush("1000 twelve 30 4242"), None);
298 }
299
300 #[test]
301 fn rates_are_per_session_then_summed() {
302 let mut c = Collector::new();
303 // Shell 1: 60 keys over 30 s = 120/min. Shell 2: 10 keys over
304 // 60 s = 10/min. Interleaved in the spool as they would be.
305 c.absorb([
306 flush(0, 0, 0, 1),
307 flush(0, 0, 0, 2),
308 flush(60_000, 10, 60, 2),
309 flush(30_000, 60, 30, 1),
310 ]);
311 let sigs = c.signals(60_000);
312 let kpm = sigs.iter().find(|s| s.name == SignalName::KeysPerMin).unwrap();
313 assert_eq!(kpm.value, Value(130.0));
314 let ss = sigs.iter().find(|s| s.name == SignalName::SessionSeconds).unwrap();
315 assert_eq!(ss.value, Value(60.0), "longest active session");
316 }
317
318 #[test]
319 fn idle_sessions_age_out() {
320 let mut c = Collector::new();
321 c.absorb([flush(0, 5, 10, 1), flush(1_000, 5, 10, 1)]);
322 assert_eq!(c.signals(1_000).len(), 2);
323 assert!(c.signals(1_000 + ACTIVE_WINDOW_MS + 1).is_empty());
324 }
325
326 #[test]
327 fn ingest_consumes_the_spool() {
328 let dir = std::env::temp_dir().join(format!("signald-spool-{}", std::process::id()));
329 std::fs::create_dir_all(&dir).unwrap();
330 let spool = dir.join("terminal.spool");
331 std::fs::write(&spool, "1000 3 1 7\nnot a record\n2000 4 2 7\n").unwrap();
332
333 let mut c = Collector::new();
334 assert_eq!(c.ingest(&spool), 2);
335 assert!(!spool.exists(), "spool is consumed, not left to grow");
336 assert!(!reading_path(&spool).exists());
337 assert_eq!(c.ingest(&spool), 0, "nothing until the hook appends again");
338
339 // State survives consumption: the rate uses both flushes.
340 let sigs = c.signals(2_000);
341 let kpm = sigs.iter().find(|s| s.name == SignalName::KeysPerMin).unwrap();
342 assert_eq!(kpm.value, Value(240.0)); // 4 keys over 1 s
343 let _ = std::fs::remove_dir_all(&dir);
344 }
345 }
346 }
347
348 /// System + hardware collector: **IOKit-only, no powermetrics, no root**
349 /// (spec §1.4). Native macOS.
350 ///
351 /// The reads live out of process in the sibling Swift package
352 /// `macos-collector/` (SwiftPM, not in this cargo workspace), which writes
353 /// `signal-schema` wire frames to stdout. This module spawns it and publishes
354 /// every frame it emits, so hardware signals reach the hub, the history
355 /// store, and subscribers by the same path as every other collector. Frames
356 /// are decoded by the same `wire::read_frame` the socket uses, so one that
357 /// fails the schema's structural checks (version, tag rule) is rejected at
358 /// this boundary.
359 pub mod hardware {
360 use std::io::{self, Read};
361 use std::path::{Path, PathBuf};
362 use std::process::{Child, Command, Stdio};
363 use std::thread;
364
365 use signal_schema::wire;
366
367 use crate::hub::Hub;
368
369 /// The collector binary's name, looked up on `PATH` when no explicit
370 /// path is configured.
371 pub const PROGRAM: &str = "macos-collector";
372
373 /// Read frames from `reader` until EOF, publishing each into `hub`.
374 /// Returns the number of frames published. A malformed frame ends the
375 /// stream with an error.
376 pub fn ingest(reader: &mut impl Read, hub: &Hub) -> io::Result<usize> {
377 let mut n = 0;
378 while let Some(sig) = wire::read_frame(reader)? {
379 hub.publish(sig);
380 n += 1;
381 }
382 Ok(n)
383 }
384
385 /// Spawn `program` streaming frames every `interval_ms` and ingest its
386 /// stdout on a background thread. Returns once the child is running;
387 /// the thread logs when the child's stream ends. The child's stdout is
388 /// a pipe, so it exits on its next write after the daemon goes away.
389 pub fn spawn(program: &Path, interval_ms: u64, hub: Hub) -> io::Result<Child> {
390 let mut child = Command::new(program)
391 .arg("--interval-ms")
392 .arg(interval_ms.to_string())
393 .stdin(Stdio::null())
394 .stdout(Stdio::piped())
395 .stderr(Stdio::inherit())
396 .spawn()?;
397 let mut stdout = child.stdout.take().expect("stdout is piped");
398 thread::spawn(move || match ingest(&mut stdout, &hub) {
399 Ok(n) => eprintln!("signald: hardware collector exited after {n} frame(s)"),
400 Err(e) => eprintln!("signald: hardware collector stream error: {e}"),
401 });
402 Ok(child)
403 }
404
405 /// Find [`PROGRAM`] on `PATH`.
406 pub fn find_on_path() -> Option<PathBuf> {
407 let path = std::env::var_os("PATH")?;
408 std::env::split_paths(&path)
409 .map(|dir| dir.join(PROGRAM))
410 .find(|p| p.is_file())
411 }
412 }
413}
414
415/// The publish side (spec §1.3): a live pub/sub fan-out with a last-value cache.
416///
417/// On connect a subscriber is handed the current value of every cached `name`
418/// immediately (so a renderer paints correct state at once), then streams
419/// updates as they change. Each subscriber runs on its own thread with its own
420/// unbounded channel, so a slow subscriber never blocks the daemon (spec §1.3:
421/// "a stuck wallpaper process must not stall the audio thread's feed").
422pub mod publish {
423 use std::io;
424 use std::io::Write;
425 use std::os::unix::net::{UnixListener, UnixStream};
426 use std::path::Path;
427 use std::thread;
428
429 use signal_schema::wire;
430
431 use crate::hub::Hub;
432
433 /// Bind `socket_path` and serve the live stream from `hub` to each
434 /// subscriber. Blocks, accepting connections until the listener errors.
435 pub fn serve(socket_path: &Path, hub: Hub) -> io::Result<()> {
436 // Clear any stale socket file from a previous run.
437 let _ = std::fs::remove_file(socket_path);
438 if let Some(parent) = socket_path.parent() {
439 std::fs::create_dir_all(parent)?;
440 }
441 let listener = UnixListener::bind(socket_path)?;
442 eprintln!("signald: listening on {}", socket_path.display());
443
444 for stream in listener.incoming() {
445 match stream {
446 Ok(stream) => {
447 let hub = hub.clone();
448 // One subscriber per thread; a misbehaving one cannot take
449 // the daemon or the other subscribers down.
450 thread::spawn(move || {
451 if let Err(e) = serve_subscriber(stream, hub) {
452 eprintln!("signald: subscriber dropped: {e}");
453 }
454 });
455 }
456 Err(e) => eprintln!("signald: accept error: {e}"),
457 }
458 }
459 Ok(())
460 }
461
462 /// Replay the last-value cache to a freshly connected subscriber, then
463 /// stream live updates until it disconnects.
464 fn serve_subscriber(mut stream: UnixStream, hub: Hub) -> io::Result<()> {
465 let (snapshot, rx) = hub.subscribe();
466 for sig in snapshot {
467 wire::write_frame(&mut stream, &sig)?;
468 }
469 stream.flush()?;
470 // `rx` yields every signal published after we subscribed. A write error
471 // means the subscriber went away; returning drops `rx`, and the hub
472 // prunes the dead sender on its next publish.
473 for sig in rx.iter() {
474 wire::write_frame(&mut stream, &sig)?;
475 stream.flush()?;
476 }
477 Ok(())
478 }
479}
480
481/// Unix millis, best-effort (spec: monotonic-corrected later).
482pub fn now_millis() -> u64 {
483 SystemTime::now()
484 .duration_since(UNIX_EPOCH)
485 .map(|d| d.as_millis() as u64)
486 .unwrap_or(0)
487}