crates/signal-schema/src/lib.rs
561 lines · 21303 bytes
1//! # signal-schema
2//!
3//! The shared, versioned wire format for the ambient-companions signal bus.
4//!
5//! ## The privacy boundary, made structural
6//!
7//! Everything in this suite is a footnote to one invariant:
8//!
9//! > No process persists, transmits, or exposes any representation from which
10//! > the content or identity of an individual keystroke, command argument, or
11//! > typed character can be recovered. Only order-free aggregates (counts,
12//! > rates, durations, codes) leave the terminal collector.
13//!
14//! This crate makes the *schema-level half* of that invariant true **by
15//! construction**: the only payload channel is [`Value`], a newtype over
16//! `f64`. There is no `text`, `bytes`, `payload`, or `content` field. A
17//! collector *cannot* emit typed content because the wire format has nowhere
18//! to put it.
19//!
20//! The two audited exceptions are non-content identifiers carried in [`Tag`]
21//! (a bundle id, an absolute repo path, or an SSH host) and are allow-listed
22//! per [`SignalName`]. Everything else is `tag == None`, enforced at the
23//! daemon boundary.
24//!
25//! The structural guarantee is not asserted in prose alone — see
26//! `tests/privacy_invariant.rs`, which fails the build if a content-carrying
27//! field is ever added.
28
29/// Bump on **any** field change to [`Signal`]. A reader skips records it does
30/// not understand rather than failing on them; see [`wire::Frame::Skipped`].
31///
32/// v3 (0.3.0): added the aggregate [`SignalName::CpuLoad`] emitted by the macOS
33/// IOKit hardware collector (`macos-collector/`, a sibling Swift package).
34///
35/// v4 (0.5.0): [`SignalName`] was cut to the metrics that have a producer and
36/// its discriminants renumbered from zero — the last version in which
37/// renumbering was possible.
38///
39/// **v5 (1.0.0) is the 1.0 contract.** Appends [`SignalName::CollectorUp`], the
40/// daemon's own health. From 1.0 onward discriminants are append-only and
41/// removing one is a breaking change.
42pub const SCHEMA_VERSION: u16 = 5;
43
44/// The collector domain a signal originated from.
45///
46/// A small closed enum — never free text.
47#[derive(Debug, Clone, Copy, PartialEq, Eq)]
48pub enum Source {
49 Terminal,
50 Git,
51 Macos,
52 Hardware,
53}
54
55impl Source {
56 /// Stable wire discriminant.
57 pub fn to_u8(self) -> u8 {
58 match self {
59 Source::Terminal => 0,
60 Source::Git => 1,
61 Source::Macos => 2,
62 Source::Hardware => 3,
63 }
64 }
65
66 /// Inverse of [`Source::to_u8`]; `None` for an unknown discriminant.
67 pub fn from_u8(v: u8) -> Option<Source> {
68 Some(match v {
69 0 => Source::Terminal,
70 1 => Source::Git,
71 2 => Source::Macos,
72 3 => Source::Hardware,
73 _ => return None,
74 })
75 }
76}
77
78/// The **enum-constrained** metric name.
79///
80/// A fixed allow-list. Unknown names are dropped at the daemon boundary. This
81/// prevents a future careless collector from inventing `last_command` as a
82/// name and shoving a string through the tag. New metrics require a new
83/// variant here (and a schema-version bump), which is a reviewed change.
84#[derive(Debug, Clone, Copy, PartialEq, Eq)]
85pub enum SignalName {
86 // --- terminal (aggregates only) ---
87 /// Keystrokes per minute, summed across active shell sessions.
88 KeysPerMin,
89 /// Seconds of the longest currently active shell session.
90 SessionSeconds,
91 // --- git ---
92 /// Commits within the collector's rolling window (last N days).
93 CommitsWindow,
94 /// Commits since local midnight.
95 CommitsToday,
96 /// Number of local branches in the repo.
97 BranchCount,
98 /// Whole days since the most recent commit on `HEAD`.
99 DaysSinceLastCommit,
100 // --- macos ---
101 /// Battery charge percentage in `[0.0, 100.0]`.
102 BatteryPct,
103 /// `1.0` charging, `0.0` not.
104 Charging,
105 /// `ProcessInfo` thermal state, `0.0` nominal through `3.0` critical.
106 ThermalState,
107 // --- hardware ---
108 /// Instantaneous battery draw in watts, from IORegistry `AppleSmartBattery`.
109 BatteryDrawW,
110 /// Aggregate CPU busy fraction across all cores in `[0.0, 1.0]`, from Mach
111 /// `host_processor_info` tick deltas.
112 CpuLoad,
113 // --- the daemon's own health (appended in v5) ---
114 /// Whether the collector named by [`Signal::source`] is running: `1.0` up,
115 /// `0.0` down. Published by the daemon's supervisor on every state change,
116 /// and once at startup so a late subscriber reads the truth from the
117 /// last-value cache.
118 ///
119 /// Liveness is not freshness. A collector can be up and still stalled; a
120 /// reader that cares should also look at how old [`Signal::ts`] is.
121 CollectorUp,
122}
123
124impl SignalName {
125 /// Whether this metric is permitted to carry a [`Tag`]. Only a small,
126 /// audited set may — everything else must have `tag == None`.
127 pub fn allows_tag(self) -> bool {
128 matches!(
129 self,
130 // All four carry a repo path, the only audited identifier left in
131 // the v4 contract.
132 SignalName::CommitsWindow
133 | SignalName::CommitsToday
134 | SignalName::BranchCount
135 | SignalName::DaysSinceLastCommit
136 )
137 }
138
139 /// Stable wire discriminant. Renumbered from zero for v4; from 1.0 onward
140 /// new metrics append here (a reviewed change that also bumps
141 /// [`SCHEMA_VERSION`]) and no discriminant is ever reused.
142 pub fn to_u8(self) -> u8 {
143 match self {
144 SignalName::KeysPerMin => 0,
145 SignalName::SessionSeconds => 1,
146 SignalName::CommitsWindow => 2,
147 SignalName::CommitsToday => 3,
148 SignalName::BranchCount => 4,
149 SignalName::DaysSinceLastCommit => 5,
150 SignalName::BatteryPct => 6,
151 SignalName::Charging => 7,
152 SignalName::ThermalState => 8,
153 SignalName::BatteryDrawW => 9,
154 SignalName::CpuLoad => 10,
155 SignalName::CollectorUp => 11,
156 }
157 }
158
159 /// Inverse of [`SignalName::to_u8`]. An unknown discriminant returns `None`
160 /// so a reader can skip names it does not understand.
161 pub fn from_u8(v: u8) -> Option<SignalName> {
162 Some(match v {
163 0 => SignalName::KeysPerMin,
164 1 => SignalName::SessionSeconds,
165 2 => SignalName::CommitsWindow,
166 3 => SignalName::CommitsToday,
167 4 => SignalName::BranchCount,
168 5 => SignalName::DaysSinceLastCommit,
169 6 => SignalName::BatteryPct,
170 7 => SignalName::Charging,
171 8 => SignalName::ThermalState,
172 9 => SignalName::BatteryDrawW,
173 10 => SignalName::CpuLoad,
174 11 => SignalName::CollectorUp,
175 _ => return None,
176 })
177 }
178}
179
180/// The **only** payload channel: a single `f64`.
181///
182/// This newtype is the load-bearing privacy primitive. There is deliberately
183/// no constructor that accepts a `String`, `&[u8]`, or any content-shaped
184/// type. If you find yourself wanting to widen this, stop: that is the
185/// privacy boundary you would be dismantling.
186#[derive(Debug, Clone, Copy, PartialEq)]
187pub struct Value(pub f64);
188
189/// A structurally validated, non-content identifier.
190///
191/// Never free text. In the v4 contract the only identifier on the wire is an
192/// absolute repo path, so [`Tag::repo_path`] is the only constructor. Adding
193/// another means a validating constructor here, a [`SignalName`] that
194/// [`SignalName::allows_tag`] permits, and a schema bump.
195#[derive(Debug, Clone, PartialEq, Eq)]
196pub struct Tag(String);
197
198impl Tag {
199 /// An absolute repo path. Validated as a shape — absolute, non-empty, no
200 /// interior NUL — a structural identifier, never free text.
201 ///
202 /// Shape is all this crate can check. Confining a tag to the roots the
203 /// daemon was told to watch needs those roots, which only the daemon
204 /// knows; `signald` enforces that as signals are collected.
205 pub fn repo_path(s: &str) -> Option<Tag> {
206 if s.starts_with('/') && !s.is_empty() && !s.contains('\0') {
207 Some(Tag(s.to_string()))
208 } else {
209 None
210 }
211 }
212
213 /// Read-only view of the validated identifier.
214 pub fn as_str(&self) -> &str {
215 &self.0
216 }
217}
218
219/// One flat, versioned record. Every field is a named scalar, a small enum, or
220/// an audited non-content identifier.
221///
222/// Note what is absent: there is no field capable of carrying a character or a
223/// string of typed content. That absence is the point.
224#[derive(Debug, Clone, PartialEq)]
225pub struct Signal {
226 /// Bump on any field change; see [`SCHEMA_VERSION`].
227 pub schema_version: u16,
228 /// Wall-clock Unix milliseconds.
229 ///
230 /// Not monotonic, and it cannot be: this value is persisted, retention
231 /// prunes on it, and `CommitsToday` means since local midnight. An NTP
232 /// step can move it backwards, so a consumer computing a rate from two
233 /// timestamps must handle a non-positive interval.
234 pub ts: u64,
235 /// Originating collector domain.
236 pub source: Source,
237 /// Enum-constrained metric name.
238 pub name: SignalName,
239 /// The only payload channel.
240 pub value: Value,
241 /// Optional, audited, non-content identifier. `Some` only for names where
242 /// [`SignalName::allows_tag`] is true.
243 pub tag: Option<Tag>,
244}
245
246impl Signal {
247 /// Validate a record against the schema's structural rules: a tag is
248 /// present only where the name allows it. This is the daemon-boundary
249 /// check; renderers can trust records that pass it.
250 pub fn is_well_formed(&self) -> bool {
251 self.schema_version == SCHEMA_VERSION
252 && (self.tag.is_none() || self.name.allows_tag())
253 }
254}
255
256/// Length-prefixed wire encoding, the framing used over the Unix socket.
257///
258/// The frame is a little-endian `u32` body length followed by the body:
259///
260/// ```text
261/// [u32 body_len] [u16 schema_version] [u64 ts] [u8 source] [u8 name]
262/// [f64 value] [u8 tag_present] [ (u16 tag_len) (tag_len bytes utf8) ]?
263/// ```
264///
265/// The payload channel is still exactly the `f64` `value` — the framing adds
266/// nowhere to put typed content. The one string on the wire is the audited
267/// `tag` identifier, and only when [`SignalName::allows_tag`] permits it.
268pub mod wire {
269 use super::{Signal, SignalName, Source, Tag, Value, SCHEMA_VERSION};
270
271 /// Encode a signal to its length-prefixed wire bytes.
272 pub fn encode(s: &Signal) -> Vec<u8> {
273 let mut body = Vec::with_capacity(24);
274 body.extend_from_slice(&s.schema_version.to_le_bytes());
275 body.extend_from_slice(&s.ts.to_le_bytes());
276 body.push(s.source.to_u8());
277 body.push(s.name.to_u8());
278 body.extend_from_slice(&s.value.0.to_le_bytes());
279 match &s.tag {
280 None => body.push(0),
281 Some(tag) => {
282 body.push(1);
283 let bytes = tag.as_str().as_bytes();
284 body.extend_from_slice(&(bytes.len() as u16).to_le_bytes());
285 body.extend_from_slice(bytes);
286 }
287 }
288
289 let mut frame = Vec::with_capacity(4 + body.len());
290 frame.extend_from_slice(&(body.len() as u32).to_le_bytes());
291 frame.extend_from_slice(&body);
292 frame
293 }
294
295 /// Decode one length-prefixed frame from the front of `buf`. Returns `None`
296 /// if the buffer is short or the frame is malformed. Trailing bytes are
297 /// ignored, so this is safe to call on a read buffer holding one frame.
298 pub fn decode(buf: &[u8]) -> Option<Signal> {
299 if buf.len() < 4 {
300 return None;
301 }
302 let body_len = u32::from_le_bytes(buf[0..4].try_into().ok()?) as usize;
303 let body = buf.get(4..4 + body_len)?;
304 decode_body(body)
305 }
306
307 /// The outcome of reading one frame.
308 ///
309 /// A renderer built against an older schema, or before a metric was
310 /// appended, must not die on the first record it does not recognise. The
311 /// length prefix makes that possible: the body can be consumed whole and
312 /// discarded, leaving the stream in sync for the next frame.
313 #[derive(Debug, Clone, PartialEq)]
314 pub enum Frame {
315 /// A record this build understands.
316 Signal(Signal),
317 /// A well-framed body this build cannot decode: a schema version it
318 /// does not know, an unassigned [`SignalName`] discriminant, or a body
319 /// that breaks the tag rule. The frame was consumed in full and the
320 /// stream is still in sync, so the caller should carry on reading.
321 Skipped,
322 /// Clean EOF at a frame boundary.
323 Eof,
324 }
325
326 /// Read exactly one frame from a stream.
327 ///
328 /// `Err` is reserved for a stream that can no longer be framed: a length
329 /// prefix that ends mid-way, or a body shorter than its prefix promised.
330 /// A body that is framed correctly but cannot be decoded is
331 /// [`Frame::Skipped`], not an error.
332 pub fn read_frame(r: &mut impl std::io::Read) -> std::io::Result<Frame> {
333 use std::io::ErrorKind;
334
335 let mut len_buf = [0u8; 4];
336 match r.read_exact(&mut len_buf) {
337 Ok(()) => {}
338 Err(e) if e.kind() == ErrorKind::UnexpectedEof => return Ok(Frame::Eof),
339 Err(e) => return Err(e),
340 }
341 let body_len = u32::from_le_bytes(len_buf) as usize;
342 let mut body = vec![0u8; body_len];
343 // A short read here means the stream is truncated: the next bytes are
344 // not a length prefix, so there is no way to resynchronise.
345 r.read_exact(&mut body)?;
346 Ok(match decode_body(&body) {
347 Some(signal) => Frame::Signal(signal),
348 None => Frame::Skipped,
349 })
350 }
351
352 /// Encode `s` and write the whole frame to a stream.
353 pub fn write_frame(w: &mut impl std::io::Write, s: &Signal) -> std::io::Result<()> {
354 w.write_all(&encode(s))
355 }
356
357 fn decode_body(body: &[u8]) -> Option<Signal> {
358 // schema_version(2) + ts(8) + source(1) + name(1) + value(8) + tag_flag(1)
359 if body.len() < 21 {
360 return None;
361 }
362 let schema_version = u16::from_le_bytes(body[0..2].try_into().ok()?);
363 let ts = u64::from_le_bytes(body[2..10].try_into().ok()?);
364 let source = Source::from_u8(body[10])?;
365 let name = SignalName::from_u8(body[11])?;
366 let value = Value(f64::from_le_bytes(body[12..20].try_into().ok()?));
367
368 let tag = match body[20] {
369 0 => None,
370 1 => {
371 let len = u16::from_le_bytes(body.get(21..23)?.try_into().ok()?) as usize;
372 let bytes = body.get(23..23 + len)?;
373 Some(Tag(std::str::from_utf8(bytes).ok()?.to_string()))
374 }
375 _ => return None,
376 };
377
378 let signal = Signal {
379 schema_version,
380 ts,
381 source,
382 name,
383 value,
384 tag,
385 };
386 // Only accept records this build understands and that obey the tag rule.
387 if signal.schema_version != SCHEMA_VERSION || !signal.is_well_formed() {
388 return None;
389 }
390 Some(signal)
391 }
392}
393
394#[cfg(test)]
395mod wire_tests {
396 use super::*;
397
398 fn sig(name: SignalName, value: f64, tag: Option<Tag>) -> Signal {
399 Signal {
400 schema_version: SCHEMA_VERSION,
401 ts: 1_723_100_000_000,
402 source: Source::Git,
403 name,
404 value: Value(value),
405 tag,
406 }
407 }
408
409 #[test]
410 fn round_trip_no_tag() {
411 let s = sig(SignalName::CommitsToday, 7.0, None);
412 let bytes = wire::encode(&s);
413 let back = wire::decode(&bytes).expect("decodes");
414 assert_eq!(s, back);
415 }
416
417 #[test]
418 fn round_trip_with_tag() {
419 let tag = Tag::repo_path("/Users/x/git/repo").expect("valid repo path");
420 let s = Signal {
421 source: Source::Git,
422 ..sig(SignalName::CommitsWindow, 3.0, Some(tag))
423 };
424 let bytes = wire::encode(&s);
425 let back = wire::decode(&bytes).expect("decodes");
426 assert_eq!(s, back);
427 }
428
429 #[test]
430 fn round_trip_preserves_float_payload() {
431 let s = sig(SignalName::DaysSinceLastCommit, 12.5, None);
432 let back = wire::decode(&wire::encode(&s)).expect("decodes");
433 assert_eq!(back.value, Value(12.5));
434 }
435
436 #[test]
437 fn stream_read_frame_round_trips_multiple() {
438 let a = sig(SignalName::CommitsWindow, 4.0, None);
439 let b = sig(SignalName::BranchCount, 2.0, None);
440 let mut buf = Vec::new();
441 wire::write_frame(&mut buf, &a).unwrap();
442 wire::write_frame(&mut buf, &b).unwrap();
443
444 let mut cursor = std::io::Cursor::new(buf);
445 let ra = wire::read_frame(&mut cursor).unwrap();
446 let rb = wire::read_frame(&mut cursor).unwrap();
447 let end = wire::read_frame(&mut cursor).unwrap();
448 assert_eq!(ra, wire::Frame::Signal(a));
449 assert_eq!(rb, wire::Frame::Signal(b));
450 assert_eq!(end, wire::Frame::Eof, "clean EOF at frame boundary");
451 }
452
453 /// Re-frame `bytes` after overwriting one body byte, so the frame stays
454 /// well-formed at the framing layer but undecodable at the schema layer.
455 fn frame_with_body_byte(s: &Signal, offset: usize, value: u8) -> Vec<u8> {
456 let mut frame = wire::encode(s);
457 frame[4 + offset] = value;
458 frame
459 }
460
461 #[test]
462 fn newer_schema_version_is_skipped_not_fatal() {
463 let good = sig(SignalName::CommitsToday, 7.0, None);
464 let mut buf = Vec::new();
465 // schema_version is the first two bytes of the body.
466 buf.extend_from_slice(&frame_with_body_byte(&good, 0, SCHEMA_VERSION as u8 + 1));
467 wire::write_frame(&mut buf, &good).unwrap();
468
469 let mut cursor = std::io::Cursor::new(buf);
470 assert_eq!(wire::read_frame(&mut cursor).unwrap(), wire::Frame::Skipped);
471 assert_eq!(
472 wire::read_frame(&mut cursor).unwrap(),
473 wire::Frame::Signal(good),
474 "the valid frame after an unknown version is still read"
475 );
476 }
477
478 #[test]
479 fn unassigned_name_discriminant_is_skipped_not_fatal() {
480 let good = sig(SignalName::CommitsToday, 7.0, None);
481 let mut buf = Vec::new();
482 // name is body byte 11; 200 is not assigned to any variant.
483 buf.extend_from_slice(&frame_with_body_byte(&good, 11, 200));
484 wire::write_frame(&mut buf, &good).unwrap();
485
486 let mut cursor = std::io::Cursor::new(buf);
487 assert_eq!(wire::read_frame(&mut cursor).unwrap(), wire::Frame::Skipped);
488 assert_eq!(
489 wire::read_frame(&mut cursor).unwrap(),
490 wire::Frame::Signal(good),
491 "the valid frame after an unknown name is still read"
492 );
493 }
494
495 #[test]
496 fn truncated_body_is_an_error_not_a_skip() {
497 let good = sig(SignalName::CommitsToday, 7.0, None);
498 let mut frame = wire::encode(&good);
499 frame.pop(); // body one byte shorter than its length prefix promises
500
501 let mut cursor = std::io::Cursor::new(frame);
502 let err = wire::read_frame(&mut cursor).expect_err("truncated frame cannot be resynced");
503 assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
504 }
505
506 #[test]
507 fn decode_rejects_short_buffer() {
508 assert!(wire::decode(&[0, 1, 2]).is_none());
509 }
510
511 /// The v5 contract: twelve names, discriminants 0..=11, each round-tripping
512 /// through the wire byte. Adding a variant without a discriminant, or
513 /// reusing one, fails here.
514 #[test]
515 fn v5_names_are_exactly_zero_through_eleven() {
516 const NAMES: [SignalName; 12] = [
517 SignalName::KeysPerMin,
518 SignalName::SessionSeconds,
519 SignalName::CommitsWindow,
520 SignalName::CommitsToday,
521 SignalName::BranchCount,
522 SignalName::DaysSinceLastCommit,
523 SignalName::BatteryPct,
524 SignalName::Charging,
525 SignalName::ThermalState,
526 SignalName::BatteryDrawW,
527 SignalName::CpuLoad,
528 SignalName::CollectorUp,
529 ];
530 for (i, name) in NAMES.iter().enumerate() {
531 assert_eq!(name.to_u8(), i as u8, "{name:?} discriminant");
532 assert_eq!(SignalName::from_u8(i as u8), Some(*name));
533 }
534 assert_eq!(SignalName::from_u8(12), None, "12 is past the frozen set");
535 }
536
537 /// Only the four git metrics may carry a tag in v4.
538 #[test]
539 fn only_git_repo_path_names_allow_a_tag() {
540 for name in [
541 SignalName::CommitsWindow,
542 SignalName::CommitsToday,
543 SignalName::BranchCount,
544 SignalName::DaysSinceLastCommit,
545 ] {
546 assert!(name.allows_tag(), "{name:?} should allow a tag");
547 }
548 for name in [
549 SignalName::KeysPerMin,
550 SignalName::SessionSeconds,
551 SignalName::BatteryPct,
552 SignalName::Charging,
553 SignalName::ThermalState,
554 SignalName::BatteryDrawW,
555 SignalName::CpuLoad,
556 SignalName::CollectorUp,
557 ] {
558 assert!(!name.allows_tag(), "{name:?} must not allow a tag");
559 }
560 }
561}