//! # signal-schema //! //! The shared, versioned wire format for the ambient-companions signal bus. //! //! ## The privacy boundary, made structural //! //! Everything in this suite is a footnote to one invariant: //! //! > No process persists, transmits, or exposes any representation from which //! > the content or identity of an individual keystroke, command argument, or //! > typed character can be recovered. Only order-free aggregates (counts, //! > rates, durations, codes) leave the terminal collector. //! //! This crate makes the *schema-level half* of that invariant true **by //! construction**: the only payload channel is [`Value`], a newtype over //! `f64`. There is no `text`, `bytes`, `payload`, or `content` field. A //! collector *cannot* emit typed content because the wire format has nowhere //! to put it. //! //! The two audited exceptions are non-content identifiers carried in [`Tag`] //! (a bundle id, an absolute repo path, or an SSH host) and are allow-listed //! per [`SignalName`]. Everything else is `tag == None`, enforced at the //! daemon boundary. //! //! The structural guarantee is not asserted in prose alone — see //! `tests/privacy_invariant.rs`, which fails the build if a content-carrying //! field is ever added. /// Bump on **any** field change to [`Signal`]. A reader skips records it does /// not understand rather than failing on them; see [`wire::Frame::Skipped`]. /// /// v3 (0.3.0): added the aggregate [`SignalName::CpuLoad`] emitted by the macOS /// IOKit hardware collector (`macos-collector/`, a sibling Swift package). /// /// v4 (0.5.0): [`SignalName`] was cut to the metrics that have a producer and /// its discriminants renumbered from zero — the last version in which /// renumbering was possible. /// /// **v5 (1.0.0) is the 1.0 contract.** Appends [`SignalName::CollectorUp`], the /// daemon's own health. From 1.0 onward discriminants are append-only and /// removing one is a breaking change. pub const SCHEMA_VERSION: u16 = 5; /// The collector domain a signal originated from. /// /// A small closed enum — never free text. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum Source { Terminal, Git, Macos, Hardware, } impl Source { /// Stable wire discriminant. pub fn to_u8(self) -> u8 { match self { Source::Terminal => 0, Source::Git => 1, Source::Macos => 2, Source::Hardware => 3, } } /// Inverse of [`Source::to_u8`]; `None` for an unknown discriminant. pub fn from_u8(v: u8) -> Option { Some(match v { 0 => Source::Terminal, 1 => Source::Git, 2 => Source::Macos, 3 => Source::Hardware, _ => return None, }) } } /// The **enum-constrained** metric name. /// /// A fixed allow-list. Unknown names are dropped at the daemon boundary. This /// prevents a future careless collector from inventing `last_command` as a /// name and shoving a string through the tag. New metrics require a new /// variant here (and a schema-version bump), which is a reviewed change. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum SignalName { // --- terminal (aggregates only) --- /// Keystrokes per minute, summed across active shell sessions. KeysPerMin, /// Seconds of the longest currently active shell session. SessionSeconds, // --- git --- /// Commits within the collector's rolling window (last N days). CommitsWindow, /// Commits since local midnight. CommitsToday, /// Number of local branches in the repo. BranchCount, /// Whole days since the most recent commit on `HEAD`. DaysSinceLastCommit, // --- macos --- /// Battery charge percentage in `[0.0, 100.0]`. BatteryPct, /// `1.0` charging, `0.0` not. Charging, /// `ProcessInfo` thermal state, `0.0` nominal through `3.0` critical. ThermalState, // --- hardware --- /// Instantaneous battery draw in watts, from IORegistry `AppleSmartBattery`. BatteryDrawW, /// Aggregate CPU busy fraction across all cores in `[0.0, 1.0]`, from Mach /// `host_processor_info` tick deltas. CpuLoad, // --- the daemon's own health (appended in v5) --- /// Whether the collector named by [`Signal::source`] is running: `1.0` up, /// `0.0` down. Published by the daemon's supervisor on every state change, /// and once at startup so a late subscriber reads the truth from the /// last-value cache. /// /// Liveness is not freshness. A collector can be up and still stalled; a /// reader that cares should also look at how old [`Signal::ts`] is. CollectorUp, } impl SignalName { /// Whether this metric is permitted to carry a [`Tag`]. Only a small, /// audited set may — everything else must have `tag == None`. pub fn allows_tag(self) -> bool { matches!( self, // All four carry a repo path, the only audited identifier left in // the v4 contract. SignalName::CommitsWindow | SignalName::CommitsToday | SignalName::BranchCount | SignalName::DaysSinceLastCommit ) } /// Stable wire discriminant. Renumbered from zero for v4; from 1.0 onward /// new metrics append here (a reviewed change that also bumps /// [`SCHEMA_VERSION`]) and no discriminant is ever reused. pub fn to_u8(self) -> u8 { match self { SignalName::KeysPerMin => 0, SignalName::SessionSeconds => 1, SignalName::CommitsWindow => 2, SignalName::CommitsToday => 3, SignalName::BranchCount => 4, SignalName::DaysSinceLastCommit => 5, SignalName::BatteryPct => 6, SignalName::Charging => 7, SignalName::ThermalState => 8, SignalName::BatteryDrawW => 9, SignalName::CpuLoad => 10, SignalName::CollectorUp => 11, } } /// Inverse of [`SignalName::to_u8`]. An unknown discriminant returns `None` /// so a reader can skip names it does not understand. pub fn from_u8(v: u8) -> Option { Some(match v { 0 => SignalName::KeysPerMin, 1 => SignalName::SessionSeconds, 2 => SignalName::CommitsWindow, 3 => SignalName::CommitsToday, 4 => SignalName::BranchCount, 5 => SignalName::DaysSinceLastCommit, 6 => SignalName::BatteryPct, 7 => SignalName::Charging, 8 => SignalName::ThermalState, 9 => SignalName::BatteryDrawW, 10 => SignalName::CpuLoad, 11 => SignalName::CollectorUp, _ => return None, }) } } /// The **only** payload channel: a single `f64`. /// /// This newtype is the load-bearing privacy primitive. There is deliberately /// no constructor that accepts a `String`, `&[u8]`, or any content-shaped /// type. If you find yourself wanting to widen this, stop: that is the /// privacy boundary you would be dismantling. #[derive(Debug, Clone, Copy, PartialEq)] pub struct Value(pub f64); /// A structurally validated, non-content identifier. /// /// Never free text. In the v4 contract the only identifier on the wire is an /// absolute repo path, so [`Tag::repo_path`] is the only constructor. Adding /// another means a validating constructor here, a [`SignalName`] that /// [`SignalName::allows_tag`] permits, and a schema bump. #[derive(Debug, Clone, PartialEq, Eq)] pub struct Tag(String); impl Tag { /// An absolute repo path. Validated as a shape — absolute, non-empty, no /// interior NUL — a structural identifier, never free text. /// /// Shape is all this crate can check. Confining a tag to the roots the /// daemon was told to watch needs those roots, which only the daemon /// knows; `signald` enforces that as signals are collected. pub fn repo_path(s: &str) -> Option { if s.starts_with('/') && !s.is_empty() && !s.contains('\0') { Some(Tag(s.to_string())) } else { None } } /// Read-only view of the validated identifier. pub fn as_str(&self) -> &str { &self.0 } } /// One flat, versioned record. Every field is a named scalar, a small enum, or /// an audited non-content identifier. /// /// Note what is absent: there is no field capable of carrying a character or a /// string of typed content. That absence is the point. #[derive(Debug, Clone, PartialEq)] pub struct Signal { /// Bump on any field change; see [`SCHEMA_VERSION`]. pub schema_version: u16, /// Wall-clock Unix milliseconds. /// /// Not monotonic, and it cannot be: this value is persisted, retention /// prunes on it, and `CommitsToday` means since local midnight. An NTP /// step can move it backwards, so a consumer computing a rate from two /// timestamps must handle a non-positive interval. pub ts: u64, /// Originating collector domain. pub source: Source, /// Enum-constrained metric name. pub name: SignalName, /// The only payload channel. pub value: Value, /// Optional, audited, non-content identifier. `Some` only for names where /// [`SignalName::allows_tag`] is true. pub tag: Option, } impl Signal { /// Validate a record against the schema's structural rules: a tag is /// present only where the name allows it. This is the daemon-boundary /// check; renderers can trust records that pass it. pub fn is_well_formed(&self) -> bool { self.schema_version == SCHEMA_VERSION && (self.tag.is_none() || self.name.allows_tag()) } } /// Length-prefixed wire encoding, the framing used over the Unix socket. /// /// The frame is a little-endian `u32` body length followed by the body: /// /// ```text /// [u32 body_len] [u16 schema_version] [u64 ts] [u8 source] [u8 name] /// [f64 value] [u8 tag_present] [ (u16 tag_len) (tag_len bytes utf8) ]? /// ``` /// /// The payload channel is still exactly the `f64` `value` — the framing adds /// nowhere to put typed content. The one string on the wire is the audited /// `tag` identifier, and only when [`SignalName::allows_tag`] permits it. pub mod wire { use super::{Signal, SignalName, Source, Tag, Value, SCHEMA_VERSION}; /// Encode a signal to its length-prefixed wire bytes. pub fn encode(s: &Signal) -> Vec { let mut body = Vec::with_capacity(24); body.extend_from_slice(&s.schema_version.to_le_bytes()); body.extend_from_slice(&s.ts.to_le_bytes()); body.push(s.source.to_u8()); body.push(s.name.to_u8()); body.extend_from_slice(&s.value.0.to_le_bytes()); match &s.tag { None => body.push(0), Some(tag) => { body.push(1); let bytes = tag.as_str().as_bytes(); body.extend_from_slice(&(bytes.len() as u16).to_le_bytes()); body.extend_from_slice(bytes); } } let mut frame = Vec::with_capacity(4 + body.len()); frame.extend_from_slice(&(body.len() as u32).to_le_bytes()); frame.extend_from_slice(&body); frame } /// Decode one length-prefixed frame from the front of `buf`. Returns `None` /// if the buffer is short or the frame is malformed. Trailing bytes are /// ignored, so this is safe to call on a read buffer holding one frame. pub fn decode(buf: &[u8]) -> Option { if buf.len() < 4 { return None; } let body_len = u32::from_le_bytes(buf[0..4].try_into().ok()?) as usize; let body = buf.get(4..4 + body_len)?; decode_body(body) } /// The outcome of reading one frame. /// /// A renderer built against an older schema, or before a metric was /// appended, must not die on the first record it does not recognise. The /// length prefix makes that possible: the body can be consumed whole and /// discarded, leaving the stream in sync for the next frame. #[derive(Debug, Clone, PartialEq)] pub enum Frame { /// A record this build understands. Signal(Signal), /// A well-framed body this build cannot decode: a schema version it /// does not know, an unassigned [`SignalName`] discriminant, or a body /// that breaks the tag rule. The frame was consumed in full and the /// stream is still in sync, so the caller should carry on reading. Skipped, /// Clean EOF at a frame boundary. Eof, } /// Read exactly one frame from a stream. /// /// `Err` is reserved for a stream that can no longer be framed: a length /// prefix that ends mid-way, or a body shorter than its prefix promised. /// A body that is framed correctly but cannot be decoded is /// [`Frame::Skipped`], not an error. pub fn read_frame(r: &mut impl std::io::Read) -> std::io::Result { use std::io::ErrorKind; let mut len_buf = [0u8; 4]; match r.read_exact(&mut len_buf) { Ok(()) => {} Err(e) if e.kind() == ErrorKind::UnexpectedEof => return Ok(Frame::Eof), Err(e) => return Err(e), } let body_len = u32::from_le_bytes(len_buf) as usize; let mut body = vec![0u8; body_len]; // A short read here means the stream is truncated: the next bytes are // not a length prefix, so there is no way to resynchronise. r.read_exact(&mut body)?; Ok(match decode_body(&body) { Some(signal) => Frame::Signal(signal), None => Frame::Skipped, }) } /// Encode `s` and write the whole frame to a stream. pub fn write_frame(w: &mut impl std::io::Write, s: &Signal) -> std::io::Result<()> { w.write_all(&encode(s)) } fn decode_body(body: &[u8]) -> Option { // schema_version(2) + ts(8) + source(1) + name(1) + value(8) + tag_flag(1) if body.len() < 21 { return None; } let schema_version = u16::from_le_bytes(body[0..2].try_into().ok()?); let ts = u64::from_le_bytes(body[2..10].try_into().ok()?); let source = Source::from_u8(body[10])?; let name = SignalName::from_u8(body[11])?; let value = Value(f64::from_le_bytes(body[12..20].try_into().ok()?)); let tag = match body[20] { 0 => None, 1 => { let len = u16::from_le_bytes(body.get(21..23)?.try_into().ok()?) as usize; let bytes = body.get(23..23 + len)?; Some(Tag(std::str::from_utf8(bytes).ok()?.to_string())) } _ => return None, }; let signal = Signal { schema_version, ts, source, name, value, tag, }; // Only accept records this build understands and that obey the tag rule. if signal.schema_version != SCHEMA_VERSION || !signal.is_well_formed() { return None; } Some(signal) } } #[cfg(test)] mod wire_tests { use super::*; fn sig(name: SignalName, value: f64, tag: Option) -> Signal { Signal { schema_version: SCHEMA_VERSION, ts: 1_723_100_000_000, source: Source::Git, name, value: Value(value), tag, } } #[test] fn round_trip_no_tag() { let s = sig(SignalName::CommitsToday, 7.0, None); let bytes = wire::encode(&s); let back = wire::decode(&bytes).expect("decodes"); assert_eq!(s, back); } #[test] fn round_trip_with_tag() { let tag = Tag::repo_path("/Users/x/git/repo").expect("valid repo path"); let s = Signal { source: Source::Git, ..sig(SignalName::CommitsWindow, 3.0, Some(tag)) }; let bytes = wire::encode(&s); let back = wire::decode(&bytes).expect("decodes"); assert_eq!(s, back); } #[test] fn round_trip_preserves_float_payload() { let s = sig(SignalName::DaysSinceLastCommit, 12.5, None); let back = wire::decode(&wire::encode(&s)).expect("decodes"); assert_eq!(back.value, Value(12.5)); } #[test] fn stream_read_frame_round_trips_multiple() { let a = sig(SignalName::CommitsWindow, 4.0, None); let b = sig(SignalName::BranchCount, 2.0, None); let mut buf = Vec::new(); wire::write_frame(&mut buf, &a).unwrap(); wire::write_frame(&mut buf, &b).unwrap(); let mut cursor = std::io::Cursor::new(buf); let ra = wire::read_frame(&mut cursor).unwrap(); let rb = wire::read_frame(&mut cursor).unwrap(); let end = wire::read_frame(&mut cursor).unwrap(); assert_eq!(ra, wire::Frame::Signal(a)); assert_eq!(rb, wire::Frame::Signal(b)); assert_eq!(end, wire::Frame::Eof, "clean EOF at frame boundary"); } /// Re-frame `bytes` after overwriting one body byte, so the frame stays /// well-formed at the framing layer but undecodable at the schema layer. fn frame_with_body_byte(s: &Signal, offset: usize, value: u8) -> Vec { let mut frame = wire::encode(s); frame[4 + offset] = value; frame } #[test] fn newer_schema_version_is_skipped_not_fatal() { let good = sig(SignalName::CommitsToday, 7.0, None); let mut buf = Vec::new(); // schema_version is the first two bytes of the body. buf.extend_from_slice(&frame_with_body_byte(&good, 0, SCHEMA_VERSION as u8 + 1)); wire::write_frame(&mut buf, &good).unwrap(); let mut cursor = std::io::Cursor::new(buf); assert_eq!(wire::read_frame(&mut cursor).unwrap(), wire::Frame::Skipped); assert_eq!( wire::read_frame(&mut cursor).unwrap(), wire::Frame::Signal(good), "the valid frame after an unknown version is still read" ); } #[test] fn unassigned_name_discriminant_is_skipped_not_fatal() { let good = sig(SignalName::CommitsToday, 7.0, None); let mut buf = Vec::new(); // name is body byte 11; 200 is not assigned to any variant. buf.extend_from_slice(&frame_with_body_byte(&good, 11, 200)); wire::write_frame(&mut buf, &good).unwrap(); let mut cursor = std::io::Cursor::new(buf); assert_eq!(wire::read_frame(&mut cursor).unwrap(), wire::Frame::Skipped); assert_eq!( wire::read_frame(&mut cursor).unwrap(), wire::Frame::Signal(good), "the valid frame after an unknown name is still read" ); } #[test] fn truncated_body_is_an_error_not_a_skip() { let good = sig(SignalName::CommitsToday, 7.0, None); let mut frame = wire::encode(&good); frame.pop(); // body one byte shorter than its length prefix promises let mut cursor = std::io::Cursor::new(frame); let err = wire::read_frame(&mut cursor).expect_err("truncated frame cannot be resynced"); assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof); } #[test] fn decode_rejects_short_buffer() { assert!(wire::decode(&[0, 1, 2]).is_none()); } /// The v5 contract: twelve names, discriminants 0..=11, each round-tripping /// through the wire byte. Adding a variant without a discriminant, or /// reusing one, fails here. #[test] fn v5_names_are_exactly_zero_through_eleven() { const NAMES: [SignalName; 12] = [ SignalName::KeysPerMin, SignalName::SessionSeconds, SignalName::CommitsWindow, SignalName::CommitsToday, SignalName::BranchCount, SignalName::DaysSinceLastCommit, SignalName::BatteryPct, SignalName::Charging, SignalName::ThermalState, SignalName::BatteryDrawW, SignalName::CpuLoad, SignalName::CollectorUp, ]; for (i, name) in NAMES.iter().enumerate() { assert_eq!(name.to_u8(), i as u8, "{name:?} discriminant"); assert_eq!(SignalName::from_u8(i as u8), Some(*name)); } assert_eq!(SignalName::from_u8(12), None, "12 is past the frozen set"); } /// Only the four git metrics may carry a tag in v4. #[test] fn only_git_repo_path_names_allow_a_tag() { for name in [ SignalName::CommitsWindow, SignalName::CommitsToday, SignalName::BranchCount, SignalName::DaysSinceLastCommit, ] { assert!(name.allows_tag(), "{name:?} should allow a tag"); } for name in [ SignalName::KeysPerMin, SignalName::SessionSeconds, SignalName::BatteryPct, SignalName::Charging, SignalName::ThermalState, SignalName::BatteryDrawW, SignalName::CpuLoad, SignalName::CollectorUp, ] { assert!(!name.allows_tag(), "{name:?} must not allow a tag"); } } }