krz/ambient-companions

Ingest macos-collector frames in signald !2

merged merged by cmc on 2026-09-04 04:50 UTC · krz/ambient-companions:feat/hw-ingest into main

6 files changed, +212 −40

README.md +18 −9
@@ -96,12 +96,15 @@ agent. Signals come from IOKit (`IOPMPowerSource` / power sources, IORegistry
9696metric can't be reached without root, it is simply **absent from the schema**
9797rather than gated behind sudo. Accessibility permission is never requested.
9898
99It ships in v0.3 as **`macos-collector/`**, a sibling **Swift** package (built
100with `swift build`, kept out of the cargo workspace — SwiftPM and cargo do not
101share a build system). It reads aggregate CPU load, battery %, charging, battery
102draw (W), and thermal state, and emits them as `signal-schema` wire frames — the
103same byte format `signald` parses. GPU/fan are deliberately omitted (no clean
104root-free IOKit channel). See `macos-collector/README.md` for the full **Swift ↔
99It ships as **`macos-collector/`**, a sibling **Swift** package (built with
100`swift build`, kept out of the cargo workspace — SwiftPM and cargo do not share
101a build system). It reads aggregate CPU load, battery %, charging, battery draw
102(W), and thermal state, and emits them as `signal-schema` wire frames — the
103same byte format `signald` parses. `signald` spawns it as a child process
104(found on `PATH`, or named with `--collector <path>`) and ingests the frames it
105writes to stdout, so hardware signals reach the hub, the history store, and
106every subscriber by the same path as git and terminal signals. GPU/fan are
107deliberately omitted (no clean root-free IOKit channel). See `macos-collector/README.md` for the full **Swift ↔
105108Rust wire contract** (the byte layout) and the shared canonical-frame test that
106109pins both sides to the same bytes.
107110
@@ -115,12 +118,13 @@ ambient-companions/
115118│ │ ├── src/lib.rs
116119│ │ └── tests/privacy_invariant.rs
117120│ ├── signald/ # the daemon (collectors + live fan-out)
118│ │ ├── src/lib.rs # git + terminal collectors, publish
121│ │ ├── src/lib.rs # git + terminal collectors, hardware ingest, publish
119122│ │ ├── src/history.rs # SQLite (WAL) history store + recent() query
120123│ │ ├── src/hub.rs # last-value cache + live fan-out
121124│ │ ├── src/main.rs # CLI, producer loop, self-attestation
122125│ │ ├── tests/git_collector.rs # aggregates vs a temp git repo
123126│ │ ├── tests/streaming.rs # last-value cache + live update
127│ │ ├── tests/hardware_ingest.rs # collector frames reach a subscriber
124128│ │ └── tests/differential_secret_typing.rs # the full privacy ship-gate
125129│ └── terminal-garden/ # first renderer: a socket subscriber
126130│ ├── src/lib.rs # signals → plots → render (unit-tested)
@@ -193,8 +197,13 @@ and — for the differential test — `zpty` modules ship with zsh).
193197 documented and pinned by a shared canonical-frame test on both sides
194198 (`crates/signal-schema/tests/hardware_wire.rs` decodes the exact bytes the
195199 Swift encoder commits to in `macos-collector/Tests/.../WireTests.swift`).
196 *Still todo:* the menu-bar permadeath pet, and the live socket handshake by
197 which `signald` ingests the collector's frames (minimal/documented in v0.3).
200 *Still todo:* the menu-bar permadeath pet.
201- **Hardware ingest. ✅ Done (v0.4).**
202 `signald` spawns `macos-collector` as a child (`--collector <path>`, or found
203 on `PATH`) and reads its stdout with the same `wire::read_frame` the socket
204 uses (`collectors::hardware`). The five hardware signals now appear in the
205 hub, the history store, and every subscriber's snapshot
206 (`crates/signald/tests/hardware_ingest.rs`).
198207- **Phase 4** — sonification (SSH-utility first, then continuous). *Out of scope.*
199208- **Phase 5** — live wallpaper (homelab, Path A) + e-ink/poster reuse.
200209 *Out of scope.*
crates/signald/src/lib.rs +63 −11
@@ -9,7 +9,8 @@
99//! **live pub/sub bus with a last-value cache** ([`hub`], [`publish`]). The
1010//! sensitive terminal collector is aggregate-only and ships behind the now-active
1111//! differential secret-typing test (see `signal-schema/tests/privacy_invariant.rs`).
12//! The system/hardware collector remains a stub (spec phase 3).
12//! v0.4 adds the **hardware path**: the daemon spawns the sibling
13//! `macos-collector` and ingests its wire frames ([`collectors::hardware`]).
1314
1415use std::time::{SystemTime, UNIX_EPOCH};
1516
@@ -205,19 +206,70 @@ pub mod collectors {
205206 }
206207 }
207208
208 use signal_schema::Signal;
209
210209 /// System + hardware collector: **IOKit-only, no powermetrics, no root**
211210 /// (spec §1.4). Native macOS.
212211 ///
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")
212 /// The reads live out of process in the sibling Swift package
213 /// `macos-collector/` (SwiftPM, not in this cargo workspace), which writes
214 /// `signal-schema` wire frames to stdout. This module spawns it and publishes
215 /// every frame it emits, so hardware signals reach the hub, the history
216 /// store, and subscribers by the same path as every other collector. Frames
217 /// are decoded by the same `wire::read_frame` the socket uses, so one that
218 /// fails the schema's structural checks (version, tag rule) is rejected at
219 /// this boundary.
220 pub mod hardware {
221 use std::io::{self, Read};
222 use std::path::{Path, PathBuf};
223 use std::process::{Child, Command, Stdio};
224 use std::thread;
225
226 use signal_schema::wire;
227
228 use crate::hub::Hub;
229
230 /// The collector binary's name, looked up on `PATH` when no explicit
231 /// path is configured.
232 pub const PROGRAM: &str = "macos-collector";
233
234 /// Read frames from `reader` until EOF, publishing each into `hub`.
235 /// Returns the number of frames published. A malformed frame ends the
236 /// stream with an error.
237 pub fn ingest(reader: &mut impl Read, hub: &Hub) -> io::Result<usize> {
238 let mut n = 0;
239 while let Some(sig) = wire::read_frame(reader)? {
240 hub.publish(sig);
241 n += 1;
242 }
243 Ok(n)
244 }
245
246 /// Spawn `program` streaming frames every `interval_ms` and ingest its
247 /// stdout on a background thread. Returns once the child is running;
248 /// the thread logs when the child's stream ends. The child's stdout is
249 /// a pipe, so it exits on its next write after the daemon goes away.
250 pub fn spawn(program: &Path, interval_ms: u64, hub: Hub) -> io::Result<Child> {
251 let mut child = Command::new(program)
252 .arg("--interval-ms")
253 .arg(interval_ms.to_string())
254 .stdin(Stdio::null())
255 .stdout(Stdio::piped())
256 .stderr(Stdio::inherit())
257 .spawn()?;
258 let mut stdout = child.stdout.take().expect("stdout is piped");
259 thread::spawn(move || match ingest(&mut stdout, &hub) {
260 Ok(n) => eprintln!("signald: hardware collector exited after {n} frame(s)"),
261 Err(e) => eprintln!("signald: hardware collector stream error: {e}"),
262 });
263 Ok(child)
264 }
265
266 /// Find [`PROGRAM`] on `PATH`.
267 pub fn find_on_path() -> Option<PathBuf> {
268 let path = std::env::var_os("PATH")?;
269 std::env::split_paths(&path)
270 .map(|dir| dir.join(PROGRAM))
271 .find(|p| p.is_file())
272 }
221273 }
222274}
223275
crates/signald/src/main.rs +31 −7
@@ -7,12 +7,14 @@
77//! ```text
88//! zsh hooks ───▶ terminal collector ─┐ (aggregate counts from the spool)
99//! git/fsevents ▶ git collector ──────┼─▶ hub: last-value cache + fan-out
10//! IOKit/AppKit ▶ system+hw collector ┘ (stub, phase 3)
10//! IOKit/AppKit ▶ system+hw collector ┘ (macos-collector child, frames on stdout)
1111//! ├─▶ publish: Unix socket (live)
1212//! └─▶ SQLite WAL history
1313//! ```
1414//!
15//! v0.2: the git and terminal collectors are live; the hub caches the latest
15//! The git and terminal collectors run in-process on a tick; the hardware
16//! collector is the sibling `macos-collector` binary, spawned as a child whose
17//! stdout frames are ingested into the same hub. The hub caches the latest
1618//! value of every metric and streams updates to subscribers; every signal is
1719//! persisted to a SQLite (WAL) history store. Collector internals live in the
1820//! `signald` library crate; this binary is argument wiring, the producer loop,
@@ -21,11 +23,13 @@
2123//! Usage:
2224//! ```text
2325//! signald [--socket <path>] [--db <path>] [--spool <path>]
24//! [--interval-ms <n>] [<repo-path> ...]
26//! [--collector <path>] [--interval-ms <n>] [<repo-path> ...]
2527//! ```
2628//! With no repo paths, the current directory is watched. The socket defaults to
2729//! `$XDG_RUNTIME_DIR/signald.sock` (fallback `~/.local/state/signald/sock`); the
28//! history db and terminal spool default alongside it.
30//! history db and terminal spool default alongside it. `--collector` names the
31//! `macos-collector` binary; by default it is looked up on `PATH` and skipped,
32//! with a log line, when absent.
2933
3034use std::path::PathBuf;
3135use std::thread;
@@ -41,13 +45,14 @@ struct Config {
4145 socket: PathBuf,
4246 db: PathBuf,
4347 spool: PathBuf,
48 collector: Option<PathBuf>,
4449 interval: Duration,
4550 repos: Vec<PathBuf>,
4651}
4752
4853fn main() {
49 print_self_attestation();
5054 let cfg = parse_args();
55 print_self_attestation(&cfg);
5156
5257 let history = match History::open(&cfg.db) {
5358 Ok(h) => {
@@ -61,6 +66,15 @@ fn main() {
6166 };
6267 let hub = Hub::with_history(history);
6368
69 // Hardware: spawn the out-of-process collector and ingest its frames. Not
70 // having one (Linux, or a dev build not on PATH) is not fatal.
71 if let Some(collector) = &cfg.collector {
72 match collectors::hardware::spawn(collector, cfg.interval.as_millis() as u64, hub.clone()) {
73 Ok(_) => eprintln!("signald: hardware collector {}", collector.display()),
74 Err(e) => eprintln!("signald: cannot start hardware collector {}: {e}", collector.display()),
75 }
76 }
77
6478 // Producer: poll the collectors on a tick and publish into the hub. Runs
6579 // for the life of the daemon, independent of any subscriber.
6680 let producer = hub.clone();
@@ -90,6 +104,7 @@ fn parse_args() -> Config {
90104 let mut socket: Option<PathBuf> = None;
91105 let mut db: Option<PathBuf> = None;
92106 let mut spool: Option<PathBuf> = None;
107 let mut collector: Option<PathBuf> = None;
93108 let mut interval_ms: u64 = 2000;
94109 let mut repos: Vec<PathBuf> = Vec::new();
95110
@@ -99,6 +114,7 @@ fn parse_args() -> Config {
99114 "--socket" => socket = args.next().map(PathBuf::from),
100115 "--db" => db = args.next().map(PathBuf::from),
101116 "--spool" => spool = args.next().map(PathBuf::from),
117 "--collector" => collector = args.next().map(PathBuf::from),
102118 "--interval-ms" => {
103119 interval_ms = args.next().and_then(|s| s.parse().ok()).unwrap_or(interval_ms)
104120 }
@@ -114,6 +130,7 @@ fn parse_args() -> Config {
114130 Config {
115131 db: db.unwrap_or_else(|| base.join("signald.sqlite")),
116132 spool: spool.unwrap_or_else(|| base.join("terminal.spool")),
133 collector: collector.or_else(collectors::hardware::find_on_path),
117134 interval: Duration::from_millis(interval_ms),
118135 repos,
119136 socket,
@@ -132,15 +149,22 @@ fn default_socket_path() -> PathBuf {
132149/// Log enabled collectors and assert none holds an input-tap capability. A real
133150/// keylogger would need one of the forbidden APIs; their absence is the point,
134151/// and this is the runtime half of that guarantee (spec §1.5).
135fn print_self_attestation() {
152fn print_self_attestation(cfg: &Config) {
136153 eprintln!("signald {} — self-attestation", env!("CARGO_PKG_VERSION"));
137154 eprintln!(" transport: unix socket (length-prefixed frames), live pub/sub");
138155 eprintln!(" history: sqlite (WAL), aggregate scalars only");
156 let hardware = match &cfg.collector {
157 Some(p) => format!("ENABLED (out-of-process {}; IOKit only, no root)", p.display()),
158 None => format!(
159 "DISABLED ({} not on PATH; pass --collector <path>)",
160 collectors::hardware::PROGRAM
161 ),
162 };
139163 for source in [Source::Terminal, Source::Git, Source::Macos, Source::Hardware] {
140164 let state = match source {
141165 Source::Git => "ENABLED (aggregate scalars only)",
142166 Source::Terminal => "ENABLED (aggregate counts from zsh spool; no input tap)",
143 Source::Macos | Source::Hardware => "stub (phase 3)",
167 Source::Macos | Source::Hardware => hardware.as_str(),
144168 };
145169 eprintln!(" collector {source:?}: input-tap capability = NONE — {state}");
146170 }
crates/signald/tests/hardware_ingest.rs added +83
@@ -0,0 +1,83 @@
1//! Hardware ingest: frames written by the out-of-process collector are read by
2//! the same `wire::read_frame` the socket uses and published into the hub, so
3//! a subscriber's snapshot carries them like any other collector's signals.
4
5use std::io::Cursor;
6
7use signal_schema::{wire, Signal, SignalName, Source, Tag, Value, SCHEMA_VERSION};
8use signald::collectors::hardware;
9use signald::hub::Hub;
10
11fn sig(source: Source, name: SignalName, value: f64) -> Signal {
12 Signal {
13 schema_version: SCHEMA_VERSION,
14 ts: 1_723_100_000_000,
15 source,
16 name,
17 value: Value(value),
18 tag: None,
19 }
20}
21
22/// The five signals `macos-collector` emits per tick, as it emits them.
23fn one_tick() -> Vec<Signal> {
24 vec![
25 sig(Source::Hardware, SignalName::CpuLoad, 0.25),
26 sig(Source::Macos, SignalName::BatteryPct, 80.0),
27 sig(Source::Macos, SignalName::Charging, 1.0),
28 sig(Source::Hardware, SignalName::BatteryDrawW, 12.5),
29 sig(Source::Macos, SignalName::ThermalState, 0.0),
30 ]
31}
32
33fn frames(signals: &[Signal]) -> Vec<u8> {
34 let mut buf = Vec::new();
35 for s in signals {
36 wire::write_frame(&mut buf, s).unwrap();
37 }
38 buf
39}
40
41#[test]
42fn canned_frames_reach_a_subscriber_snapshot() {
43 let hub = Hub::new();
44 let mut first = one_tick();
45 let mut second = one_tick();
46 second[0].value = Value(0.75); // cpu_load changes on the second tick
47 first.append(&mut second);
48
49 let n = hardware::ingest(&mut Cursor::new(frames(&first)), &hub).unwrap();
50 assert_eq!(n, 10, "every frame is published");
51
52 let (snapshot, _rx) = hub.subscribe();
53 assert_eq!(snapshot.len(), 5, "one cached value per metric");
54 let cpu = snapshot.iter().find(|s| s.name == SignalName::CpuLoad).unwrap();
55 assert_eq!(cpu.value, Value(0.75), "keep-latest");
56 assert_eq!(cpu.source, Source::Hardware);
57 for name in [
58 SignalName::BatteryPct,
59 SignalName::Charging,
60 SignalName::BatteryDrawW,
61 SignalName::ThermalState,
62 ] {
63 assert!(snapshot.iter().any(|s| s.name == name), "{name:?} missing");
64 }
65}
66
67#[test]
68fn frame_that_breaks_the_tag_rule_is_rejected_at_the_boundary() {
69 let hub = Hub::new();
70 let good = sig(Source::Hardware, SignalName::CpuLoad, 0.5);
71 // CpuLoad never allows a tag. Encode one anyway, bypassing the schema's
72 // well-formedness check, as a misbehaving collector could.
73 let mut bad = good.clone();
74 bad.tag = Some(Tag::repo_path("/not/allowed").unwrap());
75 let mut buf = frames(&[good]);
76 buf.extend(wire::encode(&bad));
77
78 let err = hardware::ingest(&mut Cursor::new(buf), &hub).unwrap_err();
79 assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
80 let snapshot = hub.snapshot();
81 assert_eq!(snapshot.len(), 1, "the good frame before it was published");
82 assert!(snapshot[0].tag.is_none());
83}
macos-collector/README.md +14 −9
@@ -97,12 +97,17 @@ execution is not required to prove they agree — the shared literal is the proo
9797(A live `swift run macos-collector --once --hex` produces frames that decode
9898byte-for-byte to this layout; see the project README for a captured run.)
9999
100## Wiring into signald (Phase 3 scope note)
101
102`signald`'s in-process `system_hw_tick` collector is a stub; this out-of-process
103Swift collector is its real realization. The **live socket handshake** by which
104`signald` ingests these frames is intentionally minimal in v0.3 (spec's stated
105Phase 3 scope): the collector emits frames to stdout / `--out` today, and the
106guarantee that `signald` parses them is the shared wire-contract test above
107(which exercises the identical `wire::decode` the socket path uses). A full
108cross-process ingest handshake is the next increment.
100## Wiring into signald
101
102`signald` spawns this binary as a child with `--interval-ms <n>` and reads the
103frames it writes to stdout (`collectors::hardware` in `crates/signald`), using
104the same `wire::read_frame` the socket path uses. It looks for
105`macos-collector` on `PATH`; a dev build is named explicitly:
106
107```sh
108signald --collector macos-collector/.build/debug/macos-collector
109```
110
111When the binary is absent (Linux, or not on `PATH`) signald logs that and runs
112without hardware signals. Stdout is a pipe, so the collector exits on its next
113write after signald goes away.
macos-collector/Sources/macos-collector/main.swift +3 −4
@@ -15,10 +15,9 @@ import CollectorCore
1515// --out <path> append raw frames to <path> (a spool the daemon can tail)
1616// instead of stdout
1717//
18// Wiring into signald: signald's live frame-ingest handshake is documented in
19// README.md and is intentionally minimal here (spec Phase 3 scope note). The
20// authoritative proof that signald parses these frames is the shared-byte
21// contract test on both sides (see WIRE section of the README).
18// Wiring into signald: signald spawns this binary as a child in the default
19// stream mode and reads the frames from its stdout. The shared-byte contract
20// test on both sides pins the format (see the WIRE section of the README).
2221
2322func fail(_ msg: String) -> Never {
2423 FileHandle.standardError.write(Data("macos-collector: \(msg)\n".utf8))