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