crates/signal-schema/src/lib.rs
465 lines · 16832 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 (spec §1.5):
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`]. Renderers negotiate on connect
30/// and ignore records they do not understand.
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).
34pub const SCHEMA_VERSION: u16 = 3;
35
36/// The collector domain a signal originated from.
37///
38/// A small closed enum — never free text.
39#[derive(Debug, Clone, Copy, PartialEq, Eq)]
40pub enum Source {
41 Terminal,
42 Git,
43 Macos,
44 Hardware,
45}
46
47impl Source {
48 /// Stable wire discriminant.
49 pub fn to_u8(self) -> u8 {
50 match self {
51 Source::Terminal => 0,
52 Source::Git => 1,
53 Source::Macos => 2,
54 Source::Hardware => 3,
55 }
56 }
57
58 /// Inverse of [`Source::to_u8`]; `None` for an unknown discriminant.
59 pub fn from_u8(v: u8) -> Option<Source> {
60 Some(match v {
61 0 => Source::Terminal,
62 1 => Source::Git,
63 2 => Source::Macos,
64 3 => Source::Hardware,
65 _ => return None,
66 })
67 }
68}
69
70/// The **enum-constrained** metric name.
71///
72/// A fixed allow-list. Unknown names are dropped at the daemon boundary. This
73/// prevents a future careless collector from inventing `last_command` as a
74/// name and shoving a string through the tag. New metrics require a new
75/// variant here (and a schema-version bump), which is a reviewed change.
76#[derive(Debug, Clone, Copy, PartialEq, Eq)]
77pub enum SignalName {
78 // --- terminal (aggregates only) ---
79 KeysPerMin,
80 InterArrivalVariance,
81 SessionSeconds,
82 CommandsPerMin,
83 ExitCodeRatio,
84 TodBucket,
85 // --- git ---
86 Commits5m,
87 LinesAdded,
88 LinesRemoved,
89 BranchLastCommitAge,
90 DirtyWorktree,
91 /// Commits within the collector's rolling window (last N days).
92 CommitsWindow,
93 /// Commits since local midnight.
94 CommitsToday,
95 /// Number of local branches in the repo.
96 BranchCount,
97 /// Whole days since the most recent commit on `HEAD`.
98 DaysSinceLastCommit,
99 // --- macos system ---
100 AppForegroundSeconds,
101 IdleSeconds,
102 BatteryPct,
103 Charging,
104 ThermalState,
105 // --- hardware ---
106 /// Aggregate CPU busy fraction across all cores in `[0.0, 1.0]`, from Mach
107 /// `host_processor_info` tick deltas. Emitted by the macOS IOKit collector
108 /// as an untagged scalar (the per-core, tagged variant is [`CpuCoreLoad`]).
109 CpuLoad,
110 CpuCoreLoad,
111 NetRxBps,
112 NetTxBps,
113 FanRpm,
114 DiskReadBps,
115 DiskWriteBps,
116 GpuUtil,
117 BatteryDrawW,
118}
119
120impl SignalName {
121 /// Whether this metric is permitted to carry a [`Tag`]. Only a small,
122 /// audited set may — everything else must have `tag == None`.
123 pub fn allows_tag(self) -> bool {
124 matches!(
125 self,
126 SignalName::BranchLastCommitAge // tag: repo/branch identifier
127 | SignalName::CommitsWindow // tag: repo path (audited, spec §1.2)
128 | SignalName::CommitsToday // tag: repo path
129 | SignalName::BranchCount // tag: repo path
130 | SignalName::DaysSinceLastCommit // tag: repo path
131 | SignalName::AppForegroundSeconds // tag: bundle id
132 | SignalName::CpuCoreLoad // tag: core index
133 )
134 }
135
136 /// Stable wire discriminant. New metrics append here (a reviewed change
137 /// that also bumps [`SCHEMA_VERSION`]).
138 pub fn to_u8(self) -> u8 {
139 match self {
140 SignalName::KeysPerMin => 0,
141 SignalName::InterArrivalVariance => 1,
142 SignalName::SessionSeconds => 2,
143 SignalName::CommandsPerMin => 3,
144 SignalName::ExitCodeRatio => 4,
145 SignalName::TodBucket => 5,
146 SignalName::Commits5m => 6,
147 SignalName::LinesAdded => 7,
148 SignalName::LinesRemoved => 8,
149 SignalName::BranchLastCommitAge => 9,
150 SignalName::DirtyWorktree => 10,
151 SignalName::CommitsWindow => 11,
152 SignalName::CommitsToday => 12,
153 SignalName::BranchCount => 13,
154 SignalName::DaysSinceLastCommit => 14,
155 SignalName::AppForegroundSeconds => 15,
156 SignalName::IdleSeconds => 16,
157 SignalName::BatteryPct => 17,
158 SignalName::Charging => 18,
159 SignalName::ThermalState => 19,
160 SignalName::CpuCoreLoad => 20,
161 SignalName::NetRxBps => 21,
162 SignalName::NetTxBps => 22,
163 SignalName::FanRpm => 23,
164 SignalName::DiskReadBps => 24,
165 SignalName::DiskWriteBps => 25,
166 SignalName::GpuUtil => 26,
167 SignalName::BatteryDrawW => 27,
168 // Appended in v3 (0.3.0); discriminants are append-only.
169 SignalName::CpuLoad => 28,
170 }
171 }
172
173 /// Inverse of [`SignalName::to_u8`]. An unknown discriminant returns `None`
174 /// so the daemon can drop names it does not understand at the boundary.
175 pub fn from_u8(v: u8) -> Option<SignalName> {
176 Some(match v {
177 0 => SignalName::KeysPerMin,
178 1 => SignalName::InterArrivalVariance,
179 2 => SignalName::SessionSeconds,
180 3 => SignalName::CommandsPerMin,
181 4 => SignalName::ExitCodeRatio,
182 5 => SignalName::TodBucket,
183 6 => SignalName::Commits5m,
184 7 => SignalName::LinesAdded,
185 8 => SignalName::LinesRemoved,
186 9 => SignalName::BranchLastCommitAge,
187 10 => SignalName::DirtyWorktree,
188 11 => SignalName::CommitsWindow,
189 12 => SignalName::CommitsToday,
190 13 => SignalName::BranchCount,
191 14 => SignalName::DaysSinceLastCommit,
192 15 => SignalName::AppForegroundSeconds,
193 16 => SignalName::IdleSeconds,
194 17 => SignalName::BatteryPct,
195 18 => SignalName::Charging,
196 19 => SignalName::ThermalState,
197 20 => SignalName::CpuCoreLoad,
198 21 => SignalName::NetRxBps,
199 22 => SignalName::NetTxBps,
200 23 => SignalName::FanRpm,
201 24 => SignalName::DiskReadBps,
202 25 => SignalName::DiskWriteBps,
203 26 => SignalName::GpuUtil,
204 27 => SignalName::BatteryDrawW,
205 28 => SignalName::CpuLoad,
206 _ => return None,
207 })
208 }
209}
210
211/// The **only** payload channel: a single `f64`.
212///
213/// This newtype is the load-bearing privacy primitive. There is deliberately
214/// no constructor that accepts a `String`, `&[u8]`, or any content-shaped
215/// type. If you find yourself wanting to widen this, stop: that is the
216/// privacy boundary you would be dismantling.
217#[derive(Debug, Clone, Copy, PartialEq)]
218pub struct Value(pub f64);
219
220/// A structurally validated, non-content identifier (spec §1.2).
221///
222/// Never free text. Constructed only via [`Tag::bundle_id`],
223/// [`Tag::repo_path`], or [`Tag::ssh_host`], each of which validates shape.
224/// M0: validation is stubbed.
225#[derive(Debug, Clone, PartialEq, Eq)]
226pub struct Tag(String);
227
228impl Tag {
229 /// A macOS bundle id, e.g. `com.apple.Terminal`. Validated against a
230 /// bundle-id shape at the daemon boundary.
231 pub fn bundle_id(_s: &str) -> Option<Tag> {
232 todo!("validate bundle-id shape (reverse-DNS); reject anything else")
233 }
234
235 /// An absolute repo path. Validated as a shape (absolute, non-empty, no
236 /// interior NUL) — a structural identifier, never free text. Per-watched-
237 /// root prefix enforcement is layered on at the daemon boundary later.
238 pub fn repo_path(s: &str) -> Option<Tag> {
239 if s.starts_with('/') && !s.is_empty() && !s.contains('\0') {
240 Some(Tag(s.to_string()))
241 } else {
242 None
243 }
244 }
245
246 /// An SSH host present in `known_hosts`.
247 pub fn ssh_host(_s: &str) -> Option<Tag> {
248 todo!("validate against ~/.ssh/known_hosts")
249 }
250
251 /// Read-only view of the validated identifier.
252 pub fn as_str(&self) -> &str {
253 &self.0
254 }
255}
256
257/// One flat, versioned record. Every field is a named scalar, a small enum, or
258/// an audited non-content identifier.
259///
260/// Note what is absent: there is no field capable of carrying a character or a
261/// string of typed content. That absence is the point.
262#[derive(Debug, Clone, PartialEq)]
263pub struct Signal {
264 /// Bump on any field change; see [`SCHEMA_VERSION`].
265 pub schema_version: u16,
266 /// Unix millis, monotonic-corrected.
267 pub ts: u64,
268 /// Originating collector domain.
269 pub source: Source,
270 /// Enum-constrained metric name.
271 pub name: SignalName,
272 /// The only payload channel.
273 pub value: Value,
274 /// Optional, audited, non-content identifier. `Some` only for names where
275 /// [`SignalName::allows_tag`] is true.
276 pub tag: Option<Tag>,
277}
278
279impl Signal {
280 /// Validate a record against the schema's structural rules: a tag is
281 /// present only where the name allows it. This is the daemon-boundary
282 /// check; renderers can trust records that pass it.
283 pub fn is_well_formed(&self) -> bool {
284 self.schema_version == SCHEMA_VERSION
285 && (self.tag.is_none() || self.name.allows_tag())
286 }
287}
288
289/// Length-prefixed wire encoding (spec §1.3: "length-prefixed framing over the
290/// Unix socket").
291///
292/// The frame is a little-endian `u32` body length followed by the body:
293///
294/// ```text
295/// [u32 body_len] [u16 schema_version] [u64 ts] [u8 source] [u8 name]
296/// [f64 value] [u8 tag_present] [ (u16 tag_len) (tag_len bytes utf8) ]?
297/// ```
298///
299/// The payload channel is still exactly the `f64` `value` — the framing adds
300/// nowhere to put typed content. The one string on the wire is the audited
301/// `tag` identifier, and only when [`SignalName::allows_tag`] permits it.
302pub mod wire {
303 use super::{Signal, SignalName, Source, Tag, Value, SCHEMA_VERSION};
304
305 /// Encode a signal to its length-prefixed wire bytes.
306 pub fn encode(s: &Signal) -> Vec<u8> {
307 let mut body = Vec::with_capacity(24);
308 body.extend_from_slice(&s.schema_version.to_le_bytes());
309 body.extend_from_slice(&s.ts.to_le_bytes());
310 body.push(s.source.to_u8());
311 body.push(s.name.to_u8());
312 body.extend_from_slice(&s.value.0.to_le_bytes());
313 match &s.tag {
314 None => body.push(0),
315 Some(tag) => {
316 body.push(1);
317 let bytes = tag.as_str().as_bytes();
318 body.extend_from_slice(&(bytes.len() as u16).to_le_bytes());
319 body.extend_from_slice(bytes);
320 }
321 }
322
323 let mut frame = Vec::with_capacity(4 + body.len());
324 frame.extend_from_slice(&(body.len() as u32).to_le_bytes());
325 frame.extend_from_slice(&body);
326 frame
327 }
328
329 /// Decode one length-prefixed frame from the front of `buf`. Returns `None`
330 /// if the buffer is short or the frame is malformed. Trailing bytes are
331 /// ignored, so this is safe to call on a read buffer holding one frame.
332 pub fn decode(buf: &[u8]) -> Option<Signal> {
333 if buf.len() < 4 {
334 return None;
335 }
336 let body_len = u32::from_le_bytes(buf[0..4].try_into().ok()?) as usize;
337 let body = buf.get(4..4 + body_len)?;
338 decode_body(body)
339 }
340
341 /// Read exactly one frame from a stream. `Ok(None)` signals a clean EOF at a
342 /// frame boundary; a partial or malformed frame is an error.
343 pub fn read_frame(r: &mut impl std::io::Read) -> std::io::Result<Option<Signal>> {
344 use std::io::{Error, ErrorKind};
345
346 let mut len_buf = [0u8; 4];
347 match r.read_exact(&mut len_buf) {
348 Ok(()) => {}
349 Err(e) if e.kind() == ErrorKind::UnexpectedEof => return Ok(None),
350 Err(e) => return Err(e),
351 }
352 let body_len = u32::from_le_bytes(len_buf) as usize;
353 let mut body = vec![0u8; body_len];
354 r.read_exact(&mut body)?;
355 decode_body(&body)
356 .map(Some)
357 .ok_or_else(|| Error::new(ErrorKind::InvalidData, "malformed signal frame"))
358 }
359
360 /// Encode `s` and write the whole frame to a stream.
361 pub fn write_frame(w: &mut impl std::io::Write, s: &Signal) -> std::io::Result<()> {
362 w.write_all(&encode(s))
363 }
364
365 fn decode_body(body: &[u8]) -> Option<Signal> {
366 // schema_version(2) + ts(8) + source(1) + name(1) + value(8) + tag_flag(1)
367 if body.len() < 21 {
368 return None;
369 }
370 let schema_version = u16::from_le_bytes(body[0..2].try_into().ok()?);
371 let ts = u64::from_le_bytes(body[2..10].try_into().ok()?);
372 let source = Source::from_u8(body[10])?;
373 let name = SignalName::from_u8(body[11])?;
374 let value = Value(f64::from_le_bytes(body[12..20].try_into().ok()?));
375
376 let tag = match body[20] {
377 0 => None,
378 1 => {
379 let len = u16::from_le_bytes(body.get(21..23)?.try_into().ok()?) as usize;
380 let bytes = body.get(23..23 + len)?;
381 Some(Tag(std::str::from_utf8(bytes).ok()?.to_string()))
382 }
383 _ => return None,
384 };
385
386 let signal = Signal {
387 schema_version,
388 ts,
389 source,
390 name,
391 value,
392 tag,
393 };
394 // Only accept records this build understands and that obey the tag rule.
395 if signal.schema_version != SCHEMA_VERSION || !signal.is_well_formed() {
396 return None;
397 }
398 Some(signal)
399 }
400}
401
402#[cfg(test)]
403mod wire_tests {
404 use super::*;
405
406 fn sig(name: SignalName, value: f64, tag: Option<Tag>) -> Signal {
407 Signal {
408 schema_version: SCHEMA_VERSION,
409 ts: 1_723_100_000_000,
410 source: Source::Git,
411 name,
412 value: Value(value),
413 tag,
414 }
415 }
416
417 #[test]
418 fn round_trip_no_tag() {
419 let s = sig(SignalName::CommitsToday, 7.0, None);
420 let bytes = wire::encode(&s);
421 let back = wire::decode(&bytes).expect("decodes");
422 assert_eq!(s, back);
423 }
424
425 #[test]
426 fn round_trip_with_tag() {
427 let tag = Tag::repo_path("/Users/x/git/repo").expect("valid repo path");
428 let s = Signal {
429 source: Source::Git,
430 ..sig(SignalName::BranchLastCommitAge, 3.0, Some(tag))
431 };
432 let bytes = wire::encode(&s);
433 let back = wire::decode(&bytes).expect("decodes");
434 assert_eq!(s, back);
435 }
436
437 #[test]
438 fn round_trip_preserves_float_payload() {
439 let s = sig(SignalName::DaysSinceLastCommit, 12.5, None);
440 let back = wire::decode(&wire::encode(&s)).expect("decodes");
441 assert_eq!(back.value, Value(12.5));
442 }
443
444 #[test]
445 fn stream_read_frame_round_trips_multiple() {
446 let a = sig(SignalName::CommitsWindow, 4.0, None);
447 let b = sig(SignalName::BranchCount, 2.0, None);
448 let mut buf = Vec::new();
449 wire::write_frame(&mut buf, &a).unwrap();
450 wire::write_frame(&mut buf, &b).unwrap();
451
452 let mut cursor = std::io::Cursor::new(buf);
453 let ra = wire::read_frame(&mut cursor).unwrap().unwrap();
454 let rb = wire::read_frame(&mut cursor).unwrap().unwrap();
455 let end = wire::read_frame(&mut cursor).unwrap();
456 assert_eq!(ra, a);
457 assert_eq!(rb, b);
458 assert!(end.is_none(), "clean EOF at frame boundary");
459 }
460
461 #[test]
462 fn decode_rejects_short_buffer() {
463 assert!(wire::decode(&[0, 1, 2]).is_none());
464 }
465}