clawft_kernel/workload_runtime/
fleet_inventory.rs

1//! Read-only inventory of Cognitum cloud-registered Seeds (card
2//! mesh-placement-23; `fleet-compat.md` sections 1b, 2(b2)).
3//!
4//! Calls the Cognitum cloud MCP's `fleet_status` tool and lists the Seeds as
5//! *candidates* for node binding ([`super::seed_bind`]). It never places,
6//! configures or registers anything through the cloud: every call passes
7//! [`ReadOnlyFleet`], which refuses any tool outside [`READ_TOOLS`] before
8//! it reaches the transport. The OAuth token lives in the operator secret
9//! store behind [`super::fleet_mcp::OAuthTokens`]; it is only ever placed
10//! in an `Authorization` header and never appears in a chain event, error
11//! or log.
12//!
13//! The parser accepts the output of `fleet_status` in the shapes its schema
14//! is expected to take (see `fixtures/cognitum_fleet_status.json`, an
15//! assumed shape until the owner captures the real one) and skips, never
16//! guesses at, an entry without a device id.
17
18use std::sync::Arc;
19
20use async_trait::async_trait;
21use serde_json::{Value, json};
22
23use super::seed::SeedApiRuntime;
24use crate::chain::{ChainManager, EVENT_KIND_WORKLOAD_REFUSE};
25
26/// Chain source of inventory events.
27pub const INVENTORY_CHAIN_SOURCE: &str = "workload.inventory";
28/// Chain kind of an inventory read.
29pub const EVENT_KIND_FLEET_INVENTORY: &str = "workload.fleet.inventory";
30/// The only MCP tools this adapter may call: reads.
31pub const READ_TOOLS: &[&str] = &["fleet_status"];
32/// Most devices accepted from one reply.
33pub const MAX_DEVICES: usize = 1000;
34
35/// Why an inventory call failed.
36#[non_exhaustive]
37#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
38pub enum FleetError {
39    /// A tool outside [`READ_TOOLS`] was asked for (never sent).
40    #[error("MCP tool {0:?} is not a read tool; the inventory adapter is read-only")]
41    NonReadTool(String),
42    /// The cloud could not be reached or answered with an error.
43    #[error("fleet transport: {0}")]
44    Transport(String),
45    /// The reply is not a recognisable `fleet_status` output.
46    #[error("fleet_status output not understood: {0}")]
47    Schema(String),
48    /// The cloud rejected the credential.
49    #[error("the Cognitum cloud refused the credential (HTTP {0})")]
50    Unauthorized(u16),
51}
52
53/// One MCP tool call transport.
54#[async_trait]
55pub trait FleetMcp: Send + Sync {
56    /// Call tool `name` with `args`; returns the tool's output as JSON.
57    async fn call_tool(&self, name: &str, args: Value) -> Result<Value, FleetError>;
58}
59
60/// Wraps a transport so only [`READ_TOOLS`] can ever be called.
61pub struct ReadOnlyFleet {
62    inner: Arc<dyn FleetMcp>,
63}
64
65impl ReadOnlyFleet {
66    /// Guard `inner`.
67    pub fn new(inner: Arc<dyn FleetMcp>) -> Self {
68        Self { inner }
69    }
70
71    /// A guard over the Cognitum cloud MCP at `url` with the operator's
72    /// OAuth token source: the only public way to reach the HTTP client.
73    pub fn http(
74        url: &str,
75        tokens: Box<dyn super::fleet_mcp::OAuthTokens>,
76    ) -> Result<Self, FleetError> {
77        Ok(Self::new(Arc::new(super::fleet_mcp::HttpFleetMcp::new(
78            url, tokens,
79        )?)))
80    }
81
82    /// Call a read tool; any other name is refused without a request.
83    pub async fn call(&self, name: &str, args: Value) -> Result<Value, FleetError> {
84        if !READ_TOOLS.contains(&name) {
85            return Err(FleetError::NonReadTool(name.to_string()));
86        }
87        self.inner.call_tool(name, args).await
88    }
89}
90
91/// A cloud-registered Seed, proposed as a node-binding candidate.
92#[derive(Debug, Clone, PartialEq, Eq)]
93pub struct FleetCandidate {
94    /// Device id.
95    pub device_id: String,
96    /// Firmware version, if reported.
97    pub firmware: Option<String>,
98    /// Online state, if reported.
99    pub online: Option<bool>,
100    /// Device public key as the cloud reports it, if reported.
101    pub device_pubkey: Option<String>,
102}
103
104fn text(v: &Value, keys: &[&str]) -> Option<String> {
105    keys.iter()
106        .find_map(|k| v.get(*k).and_then(Value::as_str))
107        .filter(|s| !s.is_empty() && s.len() <= 256)
108        .map(str::to_string)
109}
110
111fn online_of(v: &Value) -> Option<bool> {
112    if let Some(b) = v.get("online").and_then(Value::as_bool) {
113        return Some(b);
114    }
115    let s = text(v, &["status", "state"])?.to_ascii_lowercase();
116    match s.as_str() {
117        "online" | "active" | "connected" | "up" => Some(true),
118        "offline" | "inactive" | "disconnected" | "down" => Some(false),
119        _ => None,
120    }
121}
122
123/// Parse a `fleet_status` output into candidates; the second value counts
124/// entries skipped for lacking a device id.
125pub fn parse_fleet_status(out: &Value) -> Result<(Vec<FleetCandidate>, usize), FleetError> {
126    let list = match out {
127        Value::Array(a) => a,
128        Value::Object(_) => ["devices", "seeds", "fleet", "items"]
129            .iter()
130            .find_map(|k| out.get(*k).and_then(Value::as_array))
131            .ok_or_else(|| FleetError::Schema("no devices array".into()))?,
132        _ => return Err(FleetError::Schema("not an object or array".into())),
133    };
134    if list.len() > MAX_DEVICES {
135        return Err(FleetError::Schema(format!(
136            "more than {MAX_DEVICES} devices"
137        )));
138    }
139    let mut out = Vec::new();
140    let mut skipped = 0;
141    for d in list {
142        let Some(device_id) = text(d, &["device_id", "deviceId", "id"]) else {
143            skipped += 1;
144            continue;
145        };
146        out.push(FleetCandidate {
147            device_id,
148            firmware: text(
149                d,
150                &["firmware", "firmware_version", "firmwareVersion", "version"],
151            ),
152            online: online_of(d),
153            device_pubkey: text(d, &["public_key", "publicKey", "pubkey"]),
154        });
155    }
156    Ok((out, skipped))
157}
158
159/// What the local-versus-cloud key comparison found.
160#[derive(Debug, Clone, PartialEq, Eq)]
161pub enum FleetCheck {
162    /// The cloud lists this device with the same public key as local pairing.
163    Verified,
164    /// The cloud does not list the locally paired device.
165    NotListed,
166    /// The cloud lists the device but reports no key: nothing is verified.
167    KeyNotReported,
168}
169
170/// The read-only inventory adapter.
171pub struct FleetInventory {
172    fleet: ReadOnlyFleet,
173    chain: Arc<ChainManager>,
174    label: String,
175}
176
177impl FleetInventory {
178    /// Inventory over a read-only `fleet`. `label` names the credential
179    /// (never its value) in chain events.
180    pub fn new(fleet: ReadOnlyFleet, chain: Arc<ChainManager>, label: &str) -> Self {
181        Self {
182            fleet,
183            chain,
184            label: label.chars().take(64).collect(),
185        }
186    }
187
188    /// Read the fleet (optionally one `region`) and chain the read.
189    pub async fn candidates(
190        &self,
191        region: Option<&str>,
192    ) -> Result<Vec<FleetCandidate>, FleetError> {
193        let args = match region {
194            Some(r) => json!({ "region": r }),
195            None => json!({}),
196        };
197        let out = self.fleet.call("fleet_status", args).await?;
198        let (list, skipped) = parse_fleet_status(&out)?;
199        self.chain.append(
200            INVENTORY_CHAIN_SOURCE,
201            EVENT_KIND_FLEET_INVENTORY,
202            Some(json!({
203                "tool": "fleet_status", "credential": self.label, "region": region,
204                "candidates": list.iter().map(|c| json!({
205                    "device_id": c.device_id, "firmware": c.firmware, "online": c.online,
206                })).collect::<Vec<_>>(),
207                "skipped_without_device_id": skipped,
208            })),
209        );
210        Ok(list)
211    }
212
213    /// Compare the locally paired Seed (`rt`'s own `GET /api/v1/identity`)
214    /// with the cloud's listing. A different key for the same device id is
215    /// refused and chained.
216    pub async fn cross_check(
217        &self,
218        candidates: &[FleetCandidate],
219        rt: &SeedApiRuntime,
220    ) -> Result<FleetCheck, FleetError> {
221        let (device_id, key) = rt
222            .identity()
223            .await
224            .map_err(|e| FleetError::Transport(e.to_string()))?;
225        let Some(c) = candidates.iter().find(|c| c.device_id == device_id) else {
226            return Ok(FleetCheck::NotListed);
227        };
228        match &c.device_pubkey {
229            None => Ok(FleetCheck::KeyNotReported),
230            Some(k) if *k == key => Ok(FleetCheck::Verified),
231            Some(_) => {
232                let e = FleetError::Schema(format!(
233                    "the cloud's public key for {device_id} differs from the key the Seed reports locally"
234                ));
235                self.chain.append(
236                    INVENTORY_CHAIN_SOURCE,
237                    EVENT_KIND_WORKLOAD_REFUSE,
238                    Some(json!({
239                        "phase": "fleet.inventory", "code": "key_mismatch",
240                        "device_id": device_id, "node_id": rt.node_id(),
241                        "credential": self.label,
242                    })),
243                );
244                Err(e)
245            }
246        }
247    }
248}
clawft_kernel/workload_runtime/fleet_inventory.rs

clawft_kernel/workload_runtime/
fleet_inventory.rs
