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