crates/signald/src/lib.rs
296 lines · 11957 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//! The system/hardware collector remains a stub (spec phase 3).
13
14use std::time::{SystemTime, UNIX_EPOCH};
15
16pub mod history;
17pub mod hub;
18
19/// Collectors: each reduces its domain to schema scalars (spec §1.1, §1.4).
20pub mod collectors {
21 /// Git collector: shell out to `git` for counts and ages. Counts and
22 /// branch/age scalars only — never diff content. This is the cleanest
23 /// signal in the suite and the first collector wired (spec §2.2, §3).
24 pub mod git {
25 use std::path::Path;
26 use std::process::Command;
27
28 use signal_schema::{Signal, SignalName, Source, Tag, Value, SCHEMA_VERSION};
29
30 use super::super::now_millis;
31
32 /// Rolling window (days) for [`SignalName::CommitsWindow`].
33 pub const WINDOW_DAYS: u64 = 7;
34
35 /// Derive the aggregate git signals for one repo. Every value is a
36 /// scalar `f64`; the only string on the wire is the repo path carried
37 /// in the audited `tag` (spec §1.2). Returns an empty vec if `repo` is
38 /// not a git repo.
39 pub fn collect(repo: &Path) -> Vec<Signal> {
40 if !is_git_repo(repo) {
41 return Vec::new();
42 }
43 let ts = now_millis();
44 let now_secs = ts / 1000;
45 let tag = Tag::repo_path(&repo.to_string_lossy());
46
47 let mk = |name: SignalName, value: f64| Signal {
48 schema_version: SCHEMA_VERSION,
49 ts,
50 source: Source::Git,
51 name,
52 value: Value(value),
53 tag: tag.clone(),
54 };
55
56 vec![
57 mk(
58 SignalName::CommitsWindow,
59 commits_since(repo, &format!("{WINDOW_DAYS} days ago")),
60 ),
61 mk(SignalName::CommitsToday, commits_since(repo, "midnight")),
62 mk(SignalName::BranchCount, branch_count(repo)),
63 mk(
64 SignalName::DaysSinceLastCommit,
65 days_since_last_commit(repo, now_secs),
66 ),
67 ]
68 }
69
70 fn is_git_repo(repo: &Path) -> bool {
71 run_git(repo, &["rev-parse", "--is-inside-work-tree"])
72 .map(|s| s == "true")
73 .unwrap_or(false)
74 }
75
76 /// Count commits reachable from `HEAD` newer than `since` (a git
77 /// approxidate, e.g. "midnight" or "7 days ago").
78 fn commits_since(repo: &Path, since: &str) -> f64 {
79 run_git(
80 repo,
81 &["rev-list", "--count", &format!("--since={since}"), "HEAD"],
82 )
83 .and_then(|s| s.parse::<f64>().ok())
84 .unwrap_or(0.0)
85 }
86
87 fn branch_count(repo: &Path) -> f64 {
88 run_git(repo, &["for-each-ref", "--format=%(refname)", "refs/heads/"])
89 .map(|s| if s.is_empty() { 0 } else { s.lines().count() })
90 .unwrap_or(0) as f64
91 }
92
93 /// Whole days between the most recent `HEAD` commit and now. `0.0` when
94 /// the repo has no commits yet.
95 fn days_since_last_commit(repo: &Path, now_secs: u64) -> f64 {
96 match run_git(repo, &["log", "-1", "--format=%ct", "HEAD"])
97 .and_then(|s| s.parse::<u64>().ok())
98 {
99 Some(last) => (now_secs.saturating_sub(last) / 86_400) as f64,
100 None => 0.0,
101 }
102 }
103
104 /// Run `git -C <repo> <args>` and return trimmed stdout on success.
105 fn run_git(repo: &Path, args: &[&str]) -> Option<String> {
106 let out = Command::new("git")
107 .arg("-C")
108 .arg(repo)
109 .args(args)
110 .output()
111 .ok()?;
112 if !out.status.success() {
113 return None;
114 }
115 Some(String::from_utf8_lossy(&out.stdout).trim().to_string())
116 }
117 }
118
119 /// Terminal collector (the sensitive one): **no input tap, no PTY sniff.**
120 ///
121 /// The shell side (`shell-hooks/signald-hooks.zsh`) counts keystrokes with a
122 /// `zle` widget that increments a *number* and discards the key, and appends
123 /// aggregate count records — `<epoch_ms> <keys> <session_seconds>`, numbers
124 /// only — to a spool file on each `precmd`. This collector reads that spool
125 /// and derives [`SignalName::KeysPerMin`] and [`SignalName::SessionSeconds`].
126 /// It parses numbers; a line that is not three integers is dropped, so the
127 /// spool can carry nothing but counts (the privacy contract, enforced by the
128 /// differential secret-typing test).
129 pub mod terminal {
130 use std::path::Path;
131
132 use signal_schema::{Signal, SignalName, Source, Value, SCHEMA_VERSION};
133
134 use super::super::now_millis;
135
136 /// One aggregate flush parsed from the spool: a timestamp and counts.
137 /// Every field is a number — there is nowhere to put content.
138 #[derive(Debug, Clone, Copy, PartialEq)]
139 pub struct Flush {
140 pub ts_ms: u64,
141 pub keys: u64,
142 pub session_s: u64,
143 }
144
145 /// Parse one spool line. Returns `None` for anything that is not exactly
146 /// three integers, so non-count lines can never survive into a signal.
147 pub fn parse_flush(line: &str) -> Option<Flush> {
148 let mut it = line.split_whitespace();
149 let ts_ms = it.next()?.parse().ok()?;
150 let keys = it.next()?.parse().ok()?;
151 let session_s = it.next()?.parse().ok()?;
152 if it.next().is_some() {
153 return None; // extra fields => not a well-formed count record
154 }
155 Some(Flush {
156 ts_ms,
157 keys,
158 session_s,
159 })
160 }
161
162 /// Read the spool and derive the current aggregate terminal signals.
163 /// Returns an empty vec if the spool is missing or holds no count lines.
164 pub fn collect(spool: &Path) -> Vec<Signal> {
165 let text = match std::fs::read_to_string(spool) {
166 Ok(t) => t,
167 Err(_) => return Vec::new(),
168 };
169 let flushes: Vec<Flush> = text.lines().filter_map(parse_flush).collect();
170 signals_from_flushes(&flushes)
171 }
172
173 /// Pure mapping from parsed flushes to schema signals (unit-testable).
174 pub fn signals_from_flushes(flushes: &[Flush]) -> Vec<Signal> {
175 let Some(last) = flushes.last() else {
176 return Vec::new();
177 };
178 let ts = now_millis();
179 let mk = |name: SignalName, value: f64| Signal {
180 schema_version: SCHEMA_VERSION,
181 ts,
182 source: Source::Terminal,
183 name,
184 value: Value(value),
185 tag: None, // terminal aggregates are never tagged (spec §1.2)
186 };
187 vec![
188 mk(SignalName::KeysPerMin, keys_per_min(flushes)),
189 mk(SignalName::SessionSeconds, last.session_s as f64),
190 ]
191 }
192
193 /// Keys/min from the most recent flush interval; falls back to the raw
194 /// per-flush count when only one flush (or a zero interval) is available.
195 fn keys_per_min(f: &[Flush]) -> f64 {
196 if f.len() >= 2 {
197 let a = &f[f.len() - 2];
198 let b = &f[f.len() - 1];
199 let dt_ms = b.ts_ms.saturating_sub(a.ts_ms);
200 if dt_ms > 0 {
201 return b.keys as f64 * 60_000.0 / dt_ms as f64;
202 }
203 }
204 f.last().map(|x| x.keys as f64).unwrap_or(0.0)
205 }
206 }
207
208 use signal_schema::Signal;
209
210 /// System + hardware collector: **IOKit-only, no powermetrics, no root**
211 /// (spec §1.4). Native macOS.
212 ///
213 /// v0.3: the real collector ships out-of-process as the sibling Swift package
214 /// `macos-collector/` (SwiftPM, not in this cargo workspace), which reads
215 /// aggregate hardware scalars via IOKit and emits `signal-schema` wire frames
216 /// this daemon can parse (see `signal-schema/tests/hardware_wire.rs`). This
217 /// in-process hook stays a stub: signald's live frame-ingest handshake for
218 /// those frames is the next increment.
219 pub fn system_hw_tick() -> Vec<Signal> {
220 todo!("v0.3: hardware signals come from the sibling macos-collector Swift package")
221 }
222}
223
224/// The publish side (spec §1.3): a live pub/sub fan-out with a last-value cache.
225///
226/// On connect a subscriber is handed the current value of every cached `name`
227/// immediately (so a renderer paints correct state at once), then streams
228/// updates as they change. Each subscriber runs on its own thread with its own
229/// unbounded channel, so a slow subscriber never blocks the daemon (spec §1.3:
230/// "a stuck wallpaper process must not stall the audio thread's feed").
231pub mod publish {
232 use std::io;
233 use std::io::Write;
234 use std::os::unix::net::{UnixListener, UnixStream};
235 use std::path::Path;
236 use std::thread;
237
238 use signal_schema::wire;
239
240 use crate::hub::Hub;
241
242 /// Bind `socket_path` and serve the live stream from `hub` to each
243 /// subscriber. Blocks, accepting connections until the listener errors.
244 pub fn serve(socket_path: &Path, hub: Hub) -> io::Result<()> {
245 // Clear any stale socket file from a previous run.
246 let _ = std::fs::remove_file(socket_path);
247 if let Some(parent) = socket_path.parent() {
248 std::fs::create_dir_all(parent)?;
249 }
250 let listener = UnixListener::bind(socket_path)?;
251 eprintln!("signald: listening on {}", socket_path.display());
252
253 for stream in listener.incoming() {
254 match stream {
255 Ok(stream) => {
256 let hub = hub.clone();
257 // One subscriber per thread; a misbehaving one cannot take
258 // the daemon or the other subscribers down.
259 thread::spawn(move || {
260 if let Err(e) = serve_subscriber(stream, hub) {
261 eprintln!("signald: subscriber dropped: {e}");
262 }
263 });
264 }
265 Err(e) => eprintln!("signald: accept error: {e}"),
266 }
267 }
268 Ok(())
269 }
270
271 /// Replay the last-value cache to a freshly connected subscriber, then
272 /// stream live updates until it disconnects.
273 fn serve_subscriber(mut stream: UnixStream, hub: Hub) -> io::Result<()> {
274 let (snapshot, rx) = hub.subscribe();
275 for sig in snapshot {
276 wire::write_frame(&mut stream, &sig)?;
277 }
278 stream.flush()?;
279 // `rx` yields every signal published after we subscribed. A write error
280 // means the subscriber went away; returning drops `rx`, and the hub
281 // prunes the dead sender on its next publish.
282 for sig in rx.iter() {
283 wire::write_frame(&mut stream, &sig)?;
284 stream.flush()?;
285 }
286 Ok(())
287 }
288}
289
290/// Unix millis, best-effort (spec: monotonic-corrected later).
291pub fn now_millis() -> u64 {
292 SystemTime::now()
293 .duration_since(UNIX_EPOCH)
294 .map(|d| d.as_millis() as u64)
295 .unwrap_or(0)
296}