Bound the terminal spool and the history store !3

merged merged by cmc on 2026-09-04 04:50 UTC · krz/ambient-companions:fix/spool-history-bounds into main

8 files changed, +327 −81

README.md +12 −5

		@@ -181,11 +181,11 @@ and — for the differential test — `zpty` modules ship with zsh).
181	181	- Phase 2 — terminal collector. ✅ Done (v0.2).
182	182	The aggregate-only terminal path is real: `shell-hooks/signald-hooks.zsh`
183	183	counts keystrokes with a `zle` widget that increments a number and discards
184		the key, and appends `<epoch_ms> <keys> <session_seconds>` count records
185		(numbers only) to a spool; `collectors::terminal` reads the spool and derives
186		`keys_per_min` and `session_seconds`. The **differential secret-typing test is
187		active and passing** — the privacy ship-gate. *The Terminal Pet renderer is
188		still todo.*
	184	the key, and appends `<epoch_ms> <keys> <session_seconds> <session_id>` count
	185	records (numbers only) to a spool; `collectors::terminal` consumes the spool
	186	and derives `keys_per_min` and `session_seconds`. The **differential
	187	secret-typing test is active and passing** — the privacy ship-gate. *The
	188	Terminal Pet renderer is still todo.*
189	189	- Phase 3 — macOS IOKit hardware collector. ✅ Done (v0.3).
190	190	`macos-collector/` (a sibling Swift package, `swift build`) reads aggregate
191	191	hardware scalars via IOKit only — no `powermetrics`, no root: CPU load
		@@ -204,6 +204,13 @@ and — for the differential test — `zpty` modules ship with zsh).
204	204	uses (`collectors::hardware`). The five hardware signals now appear in the
205	205	hub, the history store, and every subscriber's snapshot
206	206	(`crates/signald/tests/hardware_ingest.rs`).
	207	- Bounded spool and history. ✅ Done (v0.4).
	208	The terminal spool is consumed each tick (renamed aside, read, deleted)
	209	instead of re-read in full forever. Records carry the shell's pid, so
	210	`keys_per_min` is each active shell's rate summed rather than a mix of
	211	interleaved sessions, and `session_seconds` is the longest active shell.
	212	History rows older than `--retention-days` (default 7) are pruned on open
	213	and every 1000 inserts.
207	214	- Phase 4 — sonification (SSH-utility first, then continuous). Out of scope.
208	215	- Phase 5 — live wallpaper (homelab, Path A) + e-ink/poster reuse.
209	216	Out of scope.

crates/signal-schema/tests/privacy_invariant.rs +2 −2

		@@ -205,13 +205,13 @@ zpty -d SH 2>/dev/null
205	205	let spool_bytes = std::fs::read(&spool).unwrap();
206	206	let _ = std::fs::remove_dir_all(&tmp);
207	207
208		// Parse count records (three integers per line) and confirm the typing was
	208	// Parse count records (four integers per line) and confirm the typing was
209	209	// genuinely counted — otherwise we'd be "proving" absence over an empty run.
210	210	let mut max_keys = 0u64;
211	211	let mut wire_bytes: Vec<u8> = Vec::new();
212	212	for line in String::from_utf8_lossy(&spool_bytes).lines() {
213	213	let nums: Vec<u64> = line.split_whitespace().filter_map(\|t\| t.parse().ok()).collect();
214		if nums.len() != 3 {
	214	if nums.len() != 4 {
215	215	continue; // not a well-formed count record
216	216	}
217	217	max_keys = max_keys.max(nums[1]);

crates/signald/src/history.rs +89 −10

		@@ -16,32 +16,59 @@
16	16	//! well-formed, `f64`-only signal. The differential secret-typing test writes a
17	17	//! planted secret through the collector into this store and asserts it never
18	18	//! appears here in any encoding.
	19	//!
	20	//! ## Retention
	21	//!
	22	//! The store is rolling: rows older than the retention are deleted on open and
	23	//! every [`PRUNE_EVERY`] records, so the file stays bounded for a daemon that
	24	//! runs for months.
19	25
20	26	use std::path::Path;
	27	use std::time::Duration;
21	28
22	29	use rusqlite::{params, Connection};
23	30	use signal_schema::{wire, Signal};
24	31
	32	use crate::now_millis;
	33
	34	/// Retention for [`History::open`].
	35	pub const DEFAULT_RETENTION: Duration = Duration::from_secs(7 * 86_400);
	36
	37	/// [`History::record`] prunes after this many inserts.
	38	pub const PRUNE_EVERY: u64 = 1000;
	39
25	40	/// A handle to the SQLite history database.
26	41	pub struct History {
27	42	conn: Connection,
	43	/// `None` for an in-memory store: nothing is ever pruned.
	44	retention_ms: Option<u64>,
	45	since_prune: u64,
28	46	}
29	47
30	48	impl History {
31		/// Open (creating if needed) the history database at `path` in WAL mode.
	49	/// Open (creating if needed) the history database at `path` in WAL mode
	50	/// with [`DEFAULT_RETENTION`].
32	51	pub fn open(path: &Path) -> rusqlite::Result<History> {
	52	Self::open_with_retention(path, DEFAULT_RETENTION)
	53	}
	54
	55	/// Open with an explicit retention. Rows older than it are pruned now and
	56	/// periodically as records arrive.
	57	pub fn open_with_retention(path: &Path, retention: Duration) -> rusqlite::Result<History> {
33	58	let conn = Connection::open(path)?;
34	59	// WAL: concurrent reads while the daemon writes; survives restarts.
35	60	conn.pragma_update(None, "journal_mode", "WAL")?;
36		Self::init(conn)
	61	let h = Self::init(conn, Some(retention.as_millis() as u64))?;
	62	h.prune()?;
	63	Ok(h)
37	64	}
38	65
39		/// An ephemeral in-memory store (used by tests).
	66	/// An ephemeral in-memory store (used by tests). No retention.
40	67	pub fn open_in_memory() -> rusqlite::Result<History> {
41		Self::init(Connection::open_in_memory()?)
	68	Self::init(Connection::open_in_memory()?, None)
42	69	}
43	70
44		fn init(conn: Connection) -> rusqlite::Result<History> {
	71	fn init(conn: Connection, retention_ms: Option<u64>) -> rusqlite::Result<History> {
45	72	// Columns for querying (name, ts, value, tag); `frame` is the audited
46	73	// wire encoding, reused for faithful, validated reconstruction on read.
47	74	conn.execute(
		@@ -61,11 +88,32 @@ impl History {
61	88	"CREATE INDEX IF NOT EXISTS idx_signals_name_ts ON signals(name, ts)",
62	89	[],
63	90	)?;
64		Ok(History { conn })
	91	// For the retention delete.
	92	conn.execute("CREATE INDEX IF NOT EXISTS idx_signals_ts ON signals(ts)", [])?;
	93	Ok(History {
	94	conn,
	95	retention_ms,
	96	since_prune: 0,
	97	})
65	98	}
66	99
67		/// Append one signal to the history.
68		pub fn record(&self, s: &Signal) -> rusqlite::Result<()> {
	100	/// Delete rows older than the retention. Returns the number deleted.
	101	pub fn prune(&self) -> rusqlite::Result<usize> {
	102	let Some(retention_ms) = self.retention_ms else {
	103	return Ok(0);
	104	};
	105	let cutoff = now_millis().saturating_sub(retention_ms) as i64;
	106	self.conn
	107	.execute("DELETE FROM signals WHERE ts < ?1", params![cutoff])
	108	}
	109
	110	/// Append one signal to the history, pruning every [`PRUNE_EVERY`] calls.
	111	pub fn record(&mut self, s: &Signal) -> rusqlite::Result<()> {
	112	self.since_prune += 1;
	113	if self.since_prune >= PRUNE_EVERY {
	114	self.since_prune = 0;
	115	self.prune()?;
	116	}
69	117	self.conn.execute(
70	118	"INSERT INTO signals (schema_version, ts, source, name, value, tag, frame)
71	119	VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
		@@ -141,7 +189,7 @@ mod tests {
141	189
142	190	#[test]
143	191	fn round_trip_write_then_query_recent() {
144		let h = History::open_in_memory().unwrap();
	192	let mut h = History::open_in_memory().unwrap();
145	193	let repo = Tag::repo_path("/x/repo").unwrap();
146	194	h.record(&sig(SignalName::CommitsWindow, 5.0, Some(repo.clone())))
147	195	.unwrap();
		@@ -161,7 +209,7 @@ mod tests {
161	209
162	210	#[test]
163	211	fn recent_named_filters_by_metric() {
164		let h = History::open_in_memory().unwrap();
	212	let mut h = History::open_in_memory().unwrap();
165	213	for v in [1.0, 2.0, 3.0] {
166	214	h.record(&sig(SignalName::KeysPerMin, v, None)).unwrap();
167	215	}
		@@ -173,4 +221,35 @@ mod tests {
173	221	assert!(keys.iter().all(\|s\| s.name == SignalName::KeysPerMin));
174	222	assert_eq!(keys[0].value, Value(3.0)); // newest first
175	223	}
	224
	225	#[test]
	226	fn retention_prunes_old_rows_on_open_and_periodically() {
	227	let dir = std::env::temp_dir().join(format!("signald-history-{}", std::process::id()));
	228	std::fs::create_dir_all(&dir).unwrap();
	229	let db = dir.join("h.sqlite");
	230	let retention = Duration::from_secs(3600);
	231	let ancient = Signal {
	232	ts: 1,
	233	..sig(SignalName::KeysPerMin, 1.0, None)
	234	};
	235	let fresh = Signal {
	236	ts: now_millis(),
	237	..sig(SignalName::KeysPerMin, 2.0, None)
	238	};
	239	{
	240	let mut h = History::open_with_retention(&db, retention).unwrap();
	241	h.record(&ancient).unwrap();
	242	h.record(&fresh).unwrap();
	243	assert_eq!(h.recent(10).unwrap().len(), 2);
	244	assert_eq!(h.prune().unwrap(), 1);
	245	assert_eq!(h.recent(10).unwrap().len(), 1);
	246	// Left behind for the reopen to prune.
	247	h.record(&ancient).unwrap();
	248	}
	249	let h = History::open_with_retention(&db, retention).unwrap();
	250	let rows = h.recent(10).unwrap();
	251	assert_eq!(rows.len(), 1, "reopen prunes");
	252	assert_eq!(rows[0].value, Value(2.0));
	253	let _ = std::fs::remove_dir_all(&dir);
	254	}
176	255	}

crates/signald/src/lib.rs +186 −47

		@@ -121,35 +121,50 @@ pub mod collectors {
121	121	///
122	122	/// The shell side (`shell-hooks/signald-hooks.zsh`) counts keystrokes with a
123	123	/// `zle` widget that increments a number and discards the key, and appends
124		/// aggregate count records — `<epoch_ms> <keys> <session_seconds>`, numbers
125		/// only — to a spool file on each `precmd`. This collector reads that spool
126		/// and derives [`SignalName::KeysPerMin`] and [`SignalName::SessionSeconds`].
127		/// It parses numbers; a line that is not three integers is dropped, so the
128		/// spool can carry nothing but counts (the privacy contract, enforced by the
129		/// differential secret-typing test).
	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).
130	137	pub mod terminal {
131		use std::path::Path;
	138	use std::collections::BTreeMap;
	139	use std::path::{Path, PathBuf};
132	140
133	141	use signal_schema::{Signal, SignalName, Source, Value, SCHEMA_VERSION};
134	142
135	143	use super::super::now_millis;
136	144
137		/// One aggregate flush parsed from the spool: a timestamp and counts.
138		/// Every field is a number — there is nowhere to put content.
	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.
139	152	#[derive(Debug, Clone, Copy, PartialEq)]
140	153	pub struct Flush {
141	154	pub ts_ms: u64,
142	155	pub keys: u64,
143	156	pub session_s: u64,
	157	pub session: u64,
144	158	}
145	159
146	160	/// Parse one spool line. Returns `None` for anything that is not exactly
147		/// three integers, so non-count lines can never survive into a signal.
	161	/// four integers, so non-count lines can never survive into a signal.
148	162	pub fn parse_flush(line: &str) -> Option<Flush> {
149	163	let mut it = line.split_whitespace();
150	164	let ts_ms = it.next()?.parse().ok()?;
151	165	let keys = it.next()?.parse().ok()?;
152	166	let session_s = it.next()?.parse().ok()?;
	167	let session = it.next()?.parse().ok()?;
153	168	if it.next().is_some() {
154	169	return None; // extra fields => not a well-formed count record
155	170	}
		@@ -157,52 +172,176 @@ pub mod collectors {
157	172	ts_ms,
158	173	keys,
159	174	session_s,
	175	session,
160	176	})
161	177	}
162	178
163		/// Read the spool and derive the current aggregate terminal signals.
164		/// Returns an empty vec if the spool is missing or holds no count lines.
165		pub fn collect(spool: &Path) -> Vec<Signal> {
166		let text = match std::fs::read_to_string(spool) {
167		Ok(t) => t,
168		Err(_) => return Vec::new(),
169		};
170		let flushes: Vec<Flush> = text.lines().filter_map(parse_flush).collect();
171		signals_from_flushes(&flushes)
	179	/// The previous and latest flush of one session.
	180	#[derive(Debug, Clone, Copy)]
	181	struct Session {
	182	prev: Option<Flush>,
	183	last: Flush,
172	184	}
173	185
174		/// Pure mapping from parsed flushes to schema signals (unit-testable).
175		pub fn signals_from_flushes(flushes: &[Flush]) -> Vec<Signal> {
176		let Some(last) = flushes.last() else {
177		return Vec::new();
178		};
179		let ts = now_millis();
180		let mk = \|name: SignalName, value: f64\| Signal {
181		schema_version: SCHEMA_VERSION,
182		ts,
183		source: Source::Terminal,
184		name,
185		value: Value(value),
186		tag: None, // terminal aggregates are never tagged (spec §1.2)
187		};
188		vec![
189		mk(SignalName::KeysPerMin, keys_per_min(flushes)),
190		mk(SignalName::SessionSeconds, last.session_s as f64),
191		]
	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	}
192	198	}
193	199
194		/// Keys/min from the most recent flush interval; falls back to the raw
195		/// per-flush count when only one flush (or a zero interval) is available.
196		fn keys_per_min(f: &[Flush]) -> f64 {
197		if f.len() >= 2 {
198		let a = &f[f.len() - 2];
199		let b = &f[f.len() - 1];
200		let dt_ms = b.ts_ms.saturating_sub(a.ts_ms);
201		if dt_ms > 0 {
202		return b.keys as f64 * 60_000.0 / dt_ms as f64;
	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,
203	289	}
204	290	}
205		f.last().map(\|x\| x.keys as f64).unwrap_or(0.0)
	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	}
206	345	}
207	346	}
208	347

crates/signald/src/main.rs +26 −11

		@@ -23,13 +23,15 @@
23	23	//! Usage:
24	24	//! ```text
25	25	//! signald [--socket <path>] [--db <path>] [--spool <path>]
26		//! [--collector <path>] [--interval-ms <n>] [<repo-path> ...]
	26	//! [--collector <path>] [--interval-ms <n>] [--retention-days <n>]
	27	//! [<repo-path> ...]
27	28	//! ```
28	29	//! With no repo paths, the current directory is watched. The socket defaults to
29	30	//! `$XDG_RUNTIME_DIR/signald.sock` (fallback `~/.local/state/signald/sock`); the
30	31	//! history db and terminal spool default alongside it. `--collector` names the
31	32	//! `macos-collector` binary; by default it is looked up on `PATH` and skipped,
32		//! with a log line, when absent.
	33	//! with a log line, when absent. History older than `--retention-days`
	34	//! (default 7) is pruned.
33	35
34	36	use std::path::PathBuf;
35	37	use std::thread;
		@@ -47,6 +49,7 @@ struct Config {
47	49	spool: PathBuf,
48	50	collector: Option<PathBuf>,
49	51	interval: Duration,
	52	retention: Duration,
50	53	repos: Vec<PathBuf>,
51	54	}
52	55
		@@ -54,9 +57,13 @@ fn main() {
54	57	let cfg = parse_args();
55	58	print_self_attestation(&cfg);
56	59
57		let history = match History::open(&cfg.db) {
	60	let history = match History::open_with_retention(&cfg.db, cfg.retention) {
58	61	Ok(h) => {
59		eprintln!("signald: history at {}", cfg.db.display());
	62	eprintln!(
	63	"signald: history at {} (retention {} days)",
	64	cfg.db.display(),
	65	cfg.retention.as_secs() / 86_400
	66	);
60	67	h
61	68	}
62	69	Err(e) => {
		@@ -81,16 +88,19 @@ fn main() {
81	88	let repos = cfg.repos.clone();
82	89	let spool = cfg.spool.clone();
83	90	let interval = cfg.interval;
84		thread::spawn(move \|\| loop {
85		for repo in &repos {
86		for sig in collectors::git::collect(repo) {
	91	thread::spawn(move \|\| {
	92	let mut terminal = collectors::terminal::Collector::new();
	93	loop {
	94	for repo in &repos {
	95	for sig in collectors::git::collect(repo) {
	96	producer.publish(sig);
	97	}
	98	}
	99	for sig in terminal.collect(&spool) {
87	100	producer.publish(sig);
88	101	}
	102	thread::sleep(interval);
89	103	}
90		for sig in collectors::terminal::collect(&spool) {
91		producer.publish(sig);
92		}
93		thread::sleep(interval);
94	104	});
95	105
96	106	eprintln!("signald: watching {} repo(s), spool {}", cfg.repos.len(), cfg.spool.display());
		@@ -106,6 +116,7 @@ fn parse_args() -> Config {
106	116	let mut spool: Option<PathBuf> = None;
107	117	let mut collector: Option<PathBuf> = None;
108	118	let mut interval_ms: u64 = 2000;
	119	let mut retention_days: u64 = 7;
109	120	let mut repos: Vec<PathBuf> = Vec::new();
110	121
111	122	let mut args = std::env::args().skip(1);
		@@ -118,6 +129,9 @@ fn parse_args() -> Config {
118	129	"--interval-ms" => {
119	130	interval_ms = args.next().and_then(\|s\| s.parse().ok()).unwrap_or(interval_ms)
120	131	}
	132	"--retention-days" => {
	133	retention_days = args.next().and_then(\|s\| s.parse().ok()).unwrap_or(retention_days)
	134	}
121	135	_ => repos.push(PathBuf::from(arg)),
122	136	}
123	137	}
		@@ -132,6 +146,7 @@ fn parse_args() -> Config {
132	146	spool: spool.unwrap_or_else(\|\| base.join("terminal.spool")),
133	147	collector: collector.or_else(collectors::hardware::find_on_path),
134	148	interval: Duration::from_millis(interval_ms),
	149	retention: Duration::from_secs(retention_days * 86_400),
135	150	repos,
136	151	socket,
137	152	}

crates/signald/tests/differential_secret_typing.rs +1 −1

		@@ -61,7 +61,7 @@ fn secret_typed_at_prompt_never_reaches_any_output() {
61	61	);
62	62
63	63	// 2. Run the exact daemon write path: collector -> hub(+history).
64		let signals = terminal::collect(&spool);
	64	let signals = terminal::Collector::new().collect(&spool);
65	65	assert!(!signals.is_empty(), "terminal collector produced no signals");
66	66	assert!(signals.iter().any(\|s\| s.name == SignalName::KeysPerMin));
67	67	assert!(signals.iter().any(\|s\| s.name == SignalName::SessionSeconds));

shell-hooks/README.md +5 −2

		@@ -18,8 +18,11 @@ v0.2 ships the `zle` keypress counter: a widget wraps `self-insert`, does
18	18	`(( _SIGNALD_KEYS++ ))`, then calls the built-in insert. It receives the key in
19	19	the editor and discards it — the character is never assigned to a variable that
20	20	outlives the widget and never leaves the shell. On each `precmd` the hook
21		appends one count record — `<epoch_ms> <keys> <session_seconds>`, numbers only
22		— to `$SIGNALD_SPOOL`, which `signald` reads.
	21	appends one count record — `<epoch_ms> <keys> <session_seconds> <session_id>`,
	22	numbers only — to `$SIGNALD_SPOOL`. The session id is the shell's pid, so
	23	several shells can share one spool and `signald` still derives each shell's
	24	rate separately. `signald` consumes the spool on every tick (renames it aside,
	25	reads it, deletes it), so it never grows.
23	26
24	27	This contract is enforced by the forbidden-symbol scan and the differential
25	28	secret-typing test in `crates/signal-schema/tests/privacy_invariant.rs` (plus

shell-hooks/signald-hooks.zsh +6 −3

		@@ -23,10 +23,13 @@
23	23	#
24	24	# Spool record format (all fields are NUMBERS, space-separated):
25	25	#
26		# <epoch_ms> <keys_since_last_flush> <session_seconds>
	26	# <epoch_ms> <keys_since_last_flush> <session_seconds> <session_id>
27	27	#
28	28	# One record is appended on each precmd (i.e. after each command line). There is
29		# no field capable of carrying typed content.
	29	# no field capable of carrying typed content. session_id is this shell's pid,
	30	# so the daemon keeps each shell's rate separate when several append to the
	31	# same spool. The daemon consumes the spool (renames it aside and deletes it),
	32	# which is why every write opens the file afresh with >>.
30	33	# =========================================================================
31	34
32	35	zmodload zsh/datetime 2>/dev/null
		@@ -52,7 +55,7 @@ _signald_precmd() {
52	55	local now_ms=$(( ${EPOCHREALTIME:-$EPOCHSECONDS} * 1000 ))
53	56	local session=$(( ${EPOCHSECONDS:-0} - _SIGNALD_SESSION_START ))
54	57	mkdir -p ${SIGNALD_SPOOL:h} 2>/dev/null
55		print -r -- "${now_ms%.*} ${_SIGNALD_KEYS} ${session}" >> $SIGNALD_SPOOL
	58	print -r -- "${now_ms%.*} ${_SIGNALD_KEYS} ${session} $$" >> $SIGNALD_SPOOL
56	59	_SIGNALD_KEYS=0
57	60	}
58	61