krz/ambient-companions

Add CollectorUp and key the cache by source !14

merged merged by cmc on 2026-09-04 15:56 UTC · krz/ambient-companions:feat/schema-v5-health into main

7 files changed, +80 −32

crates/signal-schema/src/lib.rs +25 −10
@@ -32,12 +32,14 @@
3232/// v3 (0.3.0): added the aggregate [`SignalName::CpuLoad`] emitted by the macOS
3333/// IOKit hardware collector (`macos-collector/`, a sibling Swift package).
3434///
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;
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;
4143
4244/// The collector domain a signal originated from.
4345///
@@ -108,6 +110,15 @@ pub enum SignalName {
108110 /// Aggregate CPU busy fraction across all cores in `[0.0, 1.0]`, from Mach
109111 /// `host_processor_info` tick deltas.
110112 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,
111122}
112123
113124impl SignalName {
@@ -141,6 +152,7 @@ impl SignalName {
141152 SignalName::ThermalState => 8,
142153 SignalName::BatteryDrawW => 9,
143154 SignalName::CpuLoad => 10,
155 SignalName::CollectorUp => 11,
144156 }
145157 }
146158
@@ -159,6 +171,7 @@ impl SignalName {
159171 8 => SignalName::ThermalState,
160172 9 => SignalName::BatteryDrawW,
161173 10 => SignalName::CpuLoad,
174 11 => SignalName::CollectorUp,
162175 _ => return None,
163176 })
164177 }
@@ -490,12 +503,12 @@ mod wire_tests {
490503 assert!(wire::decode(&[0, 1, 2]).is_none());
491504 }
492505
493 /// The v4 contract: eleven names, discriminants 0..=10, each round-tripping
506 /// The v5 contract: twelve names, discriminants 0..=11, each round-tripping
494507 /// through the wire byte. Adding a variant without a discriminant, or
495508 /// reusing one, fails here.
496509 #[test]
497 fn v4_names_are_exactly_zero_through_ten() {
498 const NAMES: [SignalName; 11] = [
510 fn v5_names_are_exactly_zero_through_eleven() {
511 const NAMES: [SignalName; 12] = [
499512 SignalName::KeysPerMin,
500513 SignalName::SessionSeconds,
501514 SignalName::CommitsWindow,
@@ -507,12 +520,13 @@ mod wire_tests {
507520 SignalName::ThermalState,
508521 SignalName::BatteryDrawW,
509522 SignalName::CpuLoad,
523 SignalName::CollectorUp,
510524 ];
511525 for (i, name) in NAMES.iter().enumerate() {
512526 assert_eq!(name.to_u8(), i as u8, "{name:?} discriminant");
513527 assert_eq!(SignalName::from_u8(i as u8), Some(*name));
514528 }
515 assert_eq!(SignalName::from_u8(11), None, "11 is past the frozen set");
529 assert_eq!(SignalName::from_u8(12), None, "12 is past the frozen set");
516530 }
517531
518532 /// Only the four git metrics may carry a tag in v4.
@@ -534,6 +548,7 @@ mod wire_tests {
534548 SignalName::ThermalState,
535549 SignalName::BatteryDrawW,
536550 SignalName::CpuLoad,
551 SignalName::CollectorUp,
537552 ] {
538553 assert!(!name.allows_tag(), "{name:?} must not allow a tag");
539554 }
crates/signal-schema/tests/hardware_wire.rs +4 −4
@@ -1,4 +1,4 @@
1//! # The Swift↔Rust hardware wire contract (v4)
1//! # The Swift↔Rust hardware wire contract (v5)
22//!
33//! The macOS IOKit collector is a sibling Swift package (`macos-collector/`)
44//! that speaks this crate's wire format. Swift and Rust are two independent
@@ -26,13 +26,13 @@ use signal_schema::{wire, Signal, SignalName, Source, Value, SCHEMA_VERSION};
2626
2727/// The canonical hardware test vector, shared verbatim with the Swift side.
2828///
29/// Signal: schema_version=4, ts=0, source=Hardware(3), name=CpuLoad(10),
29/// Signal: schema_version=5, ts=0, source=Hardware(3), name=CpuLoad(10),
3030/// value=0.5 (`f64`), tag=None.
3131///
3232/// Frame bytes (little-endian throughout):
3333/// ```text
3434/// 15 00 00 00 body_len = 21 (u32)
35/// 04 00 schema_version = 4 (u16)
35/// 05 00 schema_version = 5 (u16)
3636/// 00 00 00 00 00 00 00 00 ts = 0 (u64)
3737/// 03 source = Hardware (u8)
3838/// 0A name = CpuLoad = 10 (u8)
@@ -41,7 +41,7 @@ use signal_schema::{wire, Signal, SignalName, Source, Value, SCHEMA_VERSION};
4141/// ```
4242const CANONICAL_FRAME: [u8; 25] = [
4343 0x15, 0x00, 0x00, 0x00, // body_len = 21
44 0x04, 0x00, // schema_version = 4
44 0x05, 0x00, // schema_version = 5
4545 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // ts = 0
4646 0x03, // source = Hardware
4747 0x0A, // name = CpuLoad (10)
crates/signald/src/hub.rs +35 −7
@@ -22,10 +22,13 @@ use signal_schema::Signal;
2222
2323use crate::history::History;
2424
25/// Cache key: metric name discriminant + optional audited tag. Distinct
26/// per-repo / per-core signals coexist; a new value for the same key replaces
27/// the old one (keep-latest).
28type Key = (u8, Option<String>);
25/// Cache key: metric name discriminant, source discriminant, and optional
26/// audited tag. Distinct per-repo signals coexist, and so do two signals that
27/// share a name but come from different collectors — `SignalName::CollectorUp`
28/// is published once per `Source`, and without `source` in the key those
29/// would overwrite each other. A new value for the same key replaces the old
30/// one (keep-latest).
31type Key = (u8, u8, Option<String>);
2932
3033/// A cloneable handle to the shared hub. Clones share one cache + subscriber
3134/// set behind a mutex.
@@ -77,7 +80,11 @@ impl Hub {
7780 // A history write failing must not stop the live stream.
7881 let _ = h.record(&s);
7982 }
80 let key = (s.name.to_u8(), s.tag.as_ref().map(|t| t.as_str().to_string()));
83 let key = (
84 s.name.to_u8(),
85 s.source.to_u8(),
86 s.tag.as_ref().map(|t| t.as_str().to_string()),
87 );
8188 cache.insert(key, s.clone());
8289 subs.retain(|tx| tx.send(s.clone()).is_ok());
8390 }
@@ -138,12 +145,33 @@ mod tests {
138145 }
139146
140147 #[test]
141 fn cache_keeps_latest_per_name() {
148 fn cache_keeps_latest_per_name_and_source() {
142149 let hub = Hub::new();
143150 hub.publish(sig(SignalName::KeysPerMin, 1.0));
144151 hub.publish(sig(SignalName::KeysPerMin, 2.0));
145152 let snap = hub.snapshot();
146 assert_eq!(snap.len(), 1, "same name collapses to keep-latest");
153 assert_eq!(snap.len(), 1, "same name and source collapses to keep-latest");
147154 assert_eq!(snap[0].value, Value(2.0));
148155 }
156
157 /// `CollectorUp` is published once per source, so the cache must not treat
158 /// the name alone as the identity. Without `source` in the key the second
159 /// publish would evict the first and a subscriber would see one collector's
160 /// health standing in for all three.
161 #[test]
162 fn same_name_from_two_sources_coexists() {
163 let hub = Hub::new();
164 for (source, value) in [(Source::Git, 1.0), (Source::Hardware, 0.0)] {
165 hub.publish(Signal {
166 source,
167 ..sig(SignalName::CollectorUp, value)
168 });
169 }
170 let snap = hub.snapshot();
171 assert_eq!(snap.len(), 2, "one entry per source");
172 let git = snap.iter().find(|s| s.source == Source::Git).expect("git health");
173 let hw = snap.iter().find(|s| s.source == Source::Hardware).expect("hardware health");
174 assert_eq!(git.value, Value(1.0));
175 assert_eq!(hw.value, Value(0.0));
176 }
149177}
crates/terminal-garden/src/main.rs +9 −4
@@ -52,9 +52,10 @@ fn run(socket: &PathBuf) -> std::io::Result<()> {
5252 let stream = UnixStream::connect(socket)?;
5353 let mut reader = BufReader::new(stream);
5454
55 // Keyed by (metric name, repo tag) so per-repo signals coexist and updates
56 // replace prior values rather than accumulating.
57 let mut latest: BTreeMap<(u8, Option<String>), Signal> = BTreeMap::new();
55 // Keyed by (metric name, source, repo tag) so per-repo signals coexist,
56 // two collectors reporting the same name stay distinct, and updates replace
57 // prior values rather than accumulating.
58 let mut latest: BTreeMap<(u8, u8, Option<String>), Signal> = BTreeMap::new();
5859 loop {
5960 let sig = match wire::read_frame(&mut reader)? {
6061 wire::Frame::Signal(sig) => sig,
@@ -66,7 +67,11 @@ fn run(socket: &PathBuf) -> std::io::Result<()> {
6667 if !SUBSCRIBE.contains(&sig.name) {
6768 continue;
6869 }
69 let key = (sig.name.to_u8(), sig.tag.as_ref().map(|t| t.as_str().to_string()));
70 let key = (
71 sig.name.to_u8(),
72 sig.source.to_u8(),
73 sig.tag.as_ref().map(|t| t.as_str().to_string()),
74 );
7075 latest.insert(key, sig);
7176
7277 let signals: Vec<Signal> = latest.values().cloned().collect();
macos-collector/README.md +3 −3
@@ -62,7 +62,7 @@ multi-byte integers and the `f64` (via its IEEE-754 bit pattern) are
6262frame:
6363 [u32 body_len] little-endian length of body
6464 body:
65 [u16 schema_version] must equal 4 (v4); a reader skips other versions
65 [u16 schema_version] must equal 5 (v5); a reader skips other versions
6666 [u64 ts] unix milliseconds
6767 [u8 source] 0=terminal 1=git 2=macos 3=hardware
6868 [u8 name] SignalName discriminant (see table above)
@@ -74,12 +74,12 @@ frame:
7474
7575### Canonical frame (pinned by tests on both sides)
7676
77`schema_version=4, ts=0, source=Hardware(3), name=CpuLoad(10), value=0.5,
77`schema_version=5, ts=0, source=Hardware(3), name=CpuLoad(10), value=0.5,
7878tag=none` encodes to these exact 25 bytes:
7979
8080```text
818115 00 00 00 body_len = 21
8204 00 schema_version = 4
8205 00 schema_version = 5
838300 00 00 00 00 00 00 00 ts = 0
848403 source = Hardware
85850A name = CpuLoad (10)
macos-collector/Sources/CollectorCore/Wire.swift +2 −2
@@ -1,9 +1,9 @@
11import Foundation
22
33/// The wire schema version. **Must equal** `signal_schema::SCHEMA_VERSION` on the
4/// Rust side (v4 as of 0.5.0, the 1.0 contract). A reader skips frames whose
4/// Rust side (v5 as of 1.0.0, the 1.0 contract). A reader skips frames whose
55/// version it does not understand, so this is a hard cross-language contract.
6public let SCHEMA_VERSION: UInt16 = 4
6public let SCHEMA_VERSION: UInt16 = 5
77
88/// The collector domain. Discriminants **must match** Rust `Source::to_u8`.
99public enum Source: UInt8 {
macos-collector/Tests/CollectorCoreTests/WireTests.swift +2 −2
@@ -14,7 +14,7 @@ final class WireTests: XCTestCase {
1414 let signal = Signal(ts: 0, source: .hardware, name: .cpuLoad, value: 0.5)
1515 let expected: [UInt8] = [
1616 0x15, 0x00, 0x00, 0x00, // body_len = 21
17 0x04, 0x00, // schema_version = 4
17 0x05, 0x00, // schema_version = 5
1818 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // ts = 0
1919 0x03, // source = Hardware
2020 0x0A, // name = CpuLoad (10)
@@ -27,7 +27,7 @@ final class WireTests: XCTestCase {
2727 /// The schema version must equal the Rust `SCHEMA_VERSION`, or the daemon
2828 /// drops every frame this collector sends.
2929 func testSchemaVersionMatchesRust() {
30 XCTAssertEqual(SCHEMA_VERSION, 4)
30 XCTAssertEqual(SCHEMA_VERSION, 5)
3131 }
3232
3333 /// The tagged-frame layout matches Rust too (hardware signals are untagged,