clawft_kernel/node_facts/
probe.rs

1//! Probe orchestration and the probes shared by every OS.
2//!
3//! [`probe_capabilities`] runs every probe against a [`ProbeHost`] and
4//! returns capabilities plus notes. It never assumes: a capability is
5//! emitted only when a probe saw it (`probed`), was told about it by the
6//! operator (`claimed`), or a harness measured it (`measured`).
7
8use clawft_types::placement::{
9    AttrValue, Capability, CapabilityId, NodeFacts, ProbeNote, Provenance,
10};
11
12use super::host::ProbeHost;
13use super::{accel, linux, macos, runtimes};
14
15/// Default TTL for self-probed facts (ten minutes).
16pub const DEFAULT_FACTS_TTL_SECS: u64 = 600;
17
18/// Longest accepted container image reference.
19pub const MAX_IMAGE_REF_LEN: usize = 255;
20
21/// True for a plausible local image reference: `[A-Za-z0-9][A-Za-z0-9._/:@-]*`.
22/// Rejects a leading `-` (it would be read as an option by `docker`), spaces
23/// and any shell or control character.
24pub fn valid_image_ref(s: &str) -> bool {
25    let mut chars = s.chars();
26    s.len() <= MAX_IMAGE_REF_LEN
27        && matches!(chars.next(), Some(c) if c.is_ascii_alphanumeric())
28        && chars.all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '/' | ':' | '@' | '-'))
29}
30
31/// Remembers what the privileged emulation probe (`docker run --privileged`)
32/// listed, so it runs once per `ttl` and not on every facts refresh.
33#[derive(Debug)]
34pub struct EmulationCache {
35    ttl: std::time::Duration,
36    slot: std::sync::Mutex<Option<(std::time::Instant, String, String)>>,
37}
38
39impl EmulationCache {
40    /// Cache entries live `ttl`.
41    pub fn new(ttl: std::time::Duration) -> Self {
42        Self {
43            ttl,
44            slot: std::sync::Mutex::new(None),
45        }
46    }
47
48    /// The remembered binfmt listing for `image`, if fresh.
49    pub(crate) fn get(&self, image: &str) -> Option<String> {
50        let g = self.slot.lock().unwrap_or_else(|p| p.into_inner());
51        g.as_ref()
52            .filter(|(at, img, _)| img == image && at.elapsed() < self.ttl)
53            .map(|(_, _, l)| l.clone())
54    }
55
56    pub(crate) fn put(&self, image: &str, listing: &str) {
57        *self.slot.lock().unwrap_or_else(|p| p.into_inner()) =
58            Some((std::time::Instant::now(), image.to_string(), listing.to_string()));
59    }
60}
61
62/// What the probe is told rather than what it sees.
63#[derive(Debug, Clone)]
64pub struct ProbeConfig {
65    /// Local image used to list binfmt handlers inside a VM-backed container
66    /// engine (OrbStack, Docker Desktop). Never pulled: skipped if absent.
67    pub docker_probe_image: Option<String>,
68    /// Operator-declared sensor feeds; always advertised as `claimed`.
69    pub declared_feeds: Vec<Capability>,
70    /// Harness / admission-probe results; only `measured` `perf.*` kept.
71    pub measured: Vec<Capability>,
72    /// Model-registry output (`crate::model_manifest::model_capabilities`);
73    /// only `model.present` and a detached `store.tier.external` are kept.
74    pub models: Vec<Capability>,
75    /// Remembers the privileged emulation probe between runs (none: always run it).
76    pub emulation_cache: Option<std::sync::Arc<EmulationCache>>,
77}
78
79impl Default for ProbeConfig {
80    fn default() -> Self {
81        Self {
82            docker_probe_image: Some("alpine:3.20".into()),
83            declared_feeds: Vec::new(),
84            measured: Vec::new(),
85            models: Vec::new(),
86            emulation_cache: None,
87        }
88    }
89}
90
91/// Probe output: capabilities plus notes explaining them.
92#[derive(Debug, Clone, Default)]
93pub struct Collected {
94    /// Capabilities in emission order.
95    pub caps: Vec<Capability>,
96    /// Notes (sources, things not visible).
97    pub notes: Vec<ProbeNote>,
98}
99
100impl Collected {
101    /// Add a capability.
102    pub fn push(&mut self, cap: Capability) {
103        self.caps.push(cap);
104    }
105    /// Add a note.
106    pub fn note(&mut self, probe: &str, note: impl Into<String>) {
107        self.notes.push(ProbeNote::new(probe, note));
108    }
109    /// True if a capability with this id was emitted.
110    pub fn has(&self, id: &str) -> bool {
111        self.caps.iter().any(|c| c.id.as_str() == id)
112    }
113}
114
115/// The mount point to advertise for the internal store, or `None` when the
116/// path would carry a user or volume name (`/Users/x`, `/home/x`, `/media/x`,
117/// `/Volumes/x` ...). `/` and system paths are the same on every machine.
118pub(crate) fn public_mount(mount: &str) -> Option<&str> {
119    let first = mount.trim_start_matches('/').split('/').next().unwrap_or("");
120    const PRIVATE: &[&str] = &["Users", "home", "media", "run", "mnt", "Volumes", "root", "private"];
121    (!PRIVATE.contains(&first)).then_some(mount)
122}
123
124/// Say that `count` external volumes were seen and that their names and mount
125/// points are not advertised (they are the user's, not the mesh's).
126pub(crate) fn note_external_withheld(c: &mut Collected, count: usize) {
127    if count > 0 {
128        c.note(
129            "store.tier.external",
130            format!("{count} external volume(s) mounted; names and mount points are not advertised"),
131        );
132    }
133}
134
135/// Build a capability, or `None` if the id is not valid (for example an
136/// arch string from a tool that does not fit the id grammar).
137pub fn cap(id: &str, provenance: Provenance) -> Option<Capability> {
138    CapabilityId::new(id)
139        .ok()
140        .map(|id| Capability::new(id, provenance))
141}
142
143/// A list attribute of strings.
144pub fn str_list<S: AsRef<str>>(items: &[S]) -> AttrValue {
145    AttrValue::List(
146        items
147            .iter()
148            .map(|s| AttrValue::Str(s.as_ref().into()))
149            .collect(),
150    )
151}
152
153/// Clamp a byte count into an `Int` attribute.
154pub fn bytes_attr(n: u64) -> AttrValue {
155    AttrValue::Int(i64::try_from(n).unwrap_or(i64::MAX))
156}
157
158/// Canonical arch name for ids and `arches_*` lists.
159pub fn normalize_arch(raw: &str) -> String {
160    match raw.trim().to_ascii_lowercase().as_str() {
161        "aarch64" | "arm64" | "arm64/v8" => "aarch64".into(),
162        "arm" | "armv7" | "armv7l" | "armhf" | "arm/v7" => "armv7".into(),
163        "armv6" | "armv6l" | "arm/v6" => "armv6".into(),
164        "x86_64" | "amd64" | "x64" | "amd64/v2" | "amd64/v3" => "x86_64".into(),
165        "i386" | "i686" | "x86" | "386" => "i386".into(),
166        "mips64el" | "mips64le" => "mips64le".into(),
167        other => other
168            .chars()
169            .filter(|c| c.is_ascii_alphanumeric() || *c == '_')
170            .collect(),
171    }
172}
173
174/// Parse `df -kP <path>` output: `(available bytes, mount point)`.
175///
176/// Parsed from the right so filesystem names with spaces do not shift the
177/// columns: the capacity column is the one ending in `%`.
178pub fn parse_df(out: &str) -> Option<(u64, String)> {
179    let line = out.lines().nth(1)?;
180    let toks: Vec<&str> = line.split_whitespace().collect();
181    let pct = toks.iter().rposition(|t| t.ends_with('%'))?;
182    let avail_kb: u64 = toks.get(pct.checked_sub(1)?)?.parse().ok()?;
183    let mount = toks.get(pct + 1..)?.join(" ");
184    if mount.is_empty() {
185        return None;
186    }
187    Some((avail_kb.saturating_mul(1024), mount))
188}
189
190/// Run every probe. Order: cpu/os, memory, storage, runtimes,
191/// accelerators, formats, feeds, measured.
192pub fn probe_capabilities(host: &dyn ProbeHost, cfg: &ProbeConfig) -> Collected {
193    let mut c = Collected::default();
194    let arch = normalize_arch(&host.arch());
195    match host.os().as_str() {
196        "macos" => macos::probe(host, &arch, &mut c),
197        "linux" => linux::probe(host, &arch, &mut c),
198        other => {
199            if let Some(cp) = cap(&format!("cpu.arch.{arch}"), Provenance::Probed) {
200                c.push(cp);
201            }
202            c.note(
203                "os",
204                format!("no probes for os {other:?}; only the arch is advertised"),
205            );
206        }
207    }
208    runtimes::probe(host, &arch, cfg, &mut c);
209    accel::probe(host, &mut c);
210    derive_formats(&mut c);
211    for feed in &cfg.declared_feeds {
212        if feed.id.family() == "feed" {
213            let mut f = feed.clone();
214            f.provenance = Provenance::Claimed;
215            c.push(f);
216        }
217    }
218    if !cfg.declared_feeds.is_empty() {
219        c.note(
220            "feed",
221            "sensor feeds are operator-declared (claimed), not probed",
222        );
223    }
224    let held = c.caps.len();
225    c.caps.extend(
226        cfg.models
227            .iter()
228            .filter(|m| {
229                m.id.as_str() == "model.present" || m.id.as_str() == "store.tier.external"
230            })
231            .cloned()
232            .map(|mut m| {
233                m.provenance = Provenance::Probed;
234                m
235            }),
236    );
237    if c.caps.len() > held {
238        c.note(
239            "model",
240            "model.present is from the local model registry (existence and size checks); paths and drive names are not advertised",
241        );
242    }
243    let before = c.caps.len();
244    c.caps.extend(
245        cfg.measured
246            .iter()
247            .filter(|m| m.id.family() == "perf" && m.provenance == Provenance::Measured)
248            .cloned(),
249    );
250    if c.caps.len() > before {
251        c.note(
252            "perf",
253            format!(
254                "{} measured perf.* values from the conformance harness",
255                c.caps.len() - before
256            ),
257        );
258    }
259    c
260}
261
262/// `format.*` ids for formats an advertised runtime or accelerator
263/// handles, so a requirement can name the format directly (ADR-101).
264fn derive_formats(c: &mut Collected) {
265    // Format -> (sources, best source provenance). A format is only as
266    // trustworthy as the best runtime/accelerator that handles it: a format
267    // known only through the claimed ANE is itself claimed.
268    let mut by_format: std::collections::BTreeMap<String, (Vec<String>, Provenance)> =
269        Default::default();
270    for cp in &c.caps {
271        if !matches!(cp.id.family(), "runtime" | "accel") {
272            continue;
273        }
274        if let Some(AttrValue::List(fmts)) = cp.attrs.get("formats") {
275            for f in fmts {
276                if let AttrValue::Str(f) = f {
277                    let e = by_format
278                        .entry(f.clone())
279                        .or_insert_with(|| (Vec::new(), cp.provenance));
280                    e.0.push(cp.id.to_string());
281                    e.1 = e.1.max(cp.provenance);
282                }
283            }
284        }
285    }
286    for (fmt, (via, prov)) in by_format {
287        if let Some(fc) = cap(&format!("format.{fmt}"), prov) {
288            c.push(fc.with_attr("via", str_list(&via)));
289        }
290    }
291}
292
293/// Probe this machine and wrap the result as unsigned [`NodeFacts`].
294///
295/// Capabilities and notes beyond the NodeFacts bounds are dropped (with a
296/// note) rather than producing facts that would not validate.
297pub fn build_facts(
298    node_id: &str,
299    issued_at: u64,
300    ttl_secs: u64,
301    seq: u64,
302    mut collected: Collected,
303) -> NodeFacts {
304    use clawft_types::placement::node_facts::{MAX_FACTS_CAPABILITIES, MAX_PROBE_NOTES};
305    let mut facts = NodeFacts::new(node_id, issued_at, ttl_secs, seq);
306    collected
307        .caps
308        .retain(|c| c.validate().is_ok() && c.id.family() != "trust");
309    if collected.caps.len() > MAX_FACTS_CAPABILITIES {
310        collected.caps.truncate(MAX_FACTS_CAPABILITIES);
311        collected.note("facts", "capability list truncated to the NodeFacts bound");
312    }
313    collected.notes.truncate(MAX_PROBE_NOTES);
314    facts.capabilities = collected.caps;
315    facts.notes = collected.notes;
316    facts
317}
318
319/// Cheap live refresh of `base`: only free memory is re-read (no container
320/// engine, no subprocess beyond `vm_stat` / `/proc/meminfo`). Capability
321/// states are carried over from `base`; the caller owns busy tracking.
322pub fn refresh_live(host: &dyn ProbeHost, base: &NodeFacts, now: u64) -> NodeFacts {
323    let mut f = base.clone();
324    f.issued_at = now;
325    let free = match host.os().as_str() {
326        "macos" => macos::live_free(host),
327        "linux" => linux::live_free(host),
328        _ => None,
329    };
330    if let Some(free) = free {
331        for c in f
332            .capabilities
333            .iter_mut()
334            .filter(|c| matches!(c.id.as_str(), "mem.unified" | "mem.system"))
335        {
336            let total = match c.attrs.get("total") {
337                Some(AttrValue::Int(t)) if *t >= 0 => *t as u64,
338                _ => u64::MAX,
339            };
340            c.attrs.insert("free".into(), bytes_attr(free.min(total)));
341        }
342    }
343    f
344}
clawft_kernel/node_facts/probe.rs

clawft_kernel/node_facts/
probe.rs
