clawft_kernel/workload_ctl/
plane_seed.rs

1//! Seed nodes (ADR-100 section 5, option b): a Cognitum Seed does not run
2//! WeftOS, so there is no `workload-host` to message. The control plane
3//! addresses card 09's `remote.api` Seed adapter, held on this node, by the
4//! operator-assigned node id. Only operator-pinned store cogs go there
5//! (our signed packages go to native / container nodes).
6//!
7//! The same governance and audit apply: `workload.place` is gated and the
8//! decision chained here; the adapter's `WorkloadHost` gates and chains
9//! install / load / start / stop / unload.
10
11use std::net::SocketAddr;
12use std::sync::Arc;
13use std::time::Duration;
14
15use serde::{Deserialize, Serialize};
16use serde_json::{Value, json};
17
18use crate::chain;
19use crate::gate::GateDecision;
20use crate::workload_pkg::codec::hex_encode;
21use crate::workload_runtime::{
22    HostContract, InstanceHandle, RuntimeError, VerifiedWorkload, WorkloadConfig, WorkloadHost,
23};
24
25use clawft_types::placement::TrustTier;
26
27use super::facts::governance_tier;
28use super::host_service::CtlConfig;
29use super::msg::method;
30use super::plane::{PLANE_CHAIN_SOURCE, PlacementControlPlane, PlacementRecord, PlaneError};
31
32/// Route name recorded for Seed placements.
33pub const SEED_ROUTE: &str = "remote.api";
34
35/// What the controller keeps for one instance placed on a Seed: the
36/// adapter handle plus the store pin it came from, so the instance can be
37/// re-adopted by a fresh adapter after a controller restart.
38#[derive(Debug, Clone, Serialize, Deserialize)]
39pub(super) struct SeedEntry {
40    pub handle: InstanceHandle,
41    pub registry: String,
42    pub version: String,
43    #[serde(default)]
44    pub sha256: Option<String>,
45}
46
47/// An operator store pin to place on a Seed.
48#[derive(Debug, Clone, Serialize, Deserialize)]
49pub struct StorePinOrder {
50    /// Operator-assigned Seed node id.
51    pub node_id: String,
52    /// Store registry (`cognitum`).
53    pub registry: String,
54    /// Cog id.
55    pub id: String,
56    /// Pinned version.
57    pub version: String,
58    /// Optional expected SHA-256 of the store binary.
59    #[serde(default)]
60    pub sha256: Option<String>,
61    /// Instance configuration.
62    pub config: CtlConfig,
63    /// Start after installing (Seeds auto-start; the adapter stops first).
64    #[serde(default)]
65    pub start: bool,
66}
67
68impl PlacementControlPlane {
69    /// Register a Seed adapter under its operator-assigned node id and
70    /// trust tier (governance sees that tier, not a default).
71    pub fn add_seed(
72        &self,
73        node_id: &str,
74        host: Arc<WorkloadHost>,
75        tier: TrustTier,
76    ) -> Result<(), PlaneError> {
77        if !crate::workload_pkg::manifest::valid_token(node_id, 64) {
78            return Err(PlaneError::Invalid(
79                "seed node id must be a plain token".into(),
80            ));
81        }
82        // A remote.api node must be reached over a pinned link (or an
83        // explicit lab opt-in); otherwise it is not registered at all.
84        host.runtime()
85            .link_security()
86            .require_pinned(node_id)
87            .map_err(PlaneError::Invalid)?;
88        host.set_node_tier(governance_tier(tier));
89        if let Ok(mut s) = self.seeds.write() {
90            s.insert(node_id.to_string(), (host, tier));
91        }
92        Ok(())
93    }
94
95    fn seed(&self, node_id: &str) -> Option<(Arc<WorkloadHost>, TrustTier)> {
96        self.seeds.read().ok().and_then(|s| s.get(node_id).cloned())
97    }
98
99    fn chain_here(&self, kind: &str, payload: Value) -> String {
100        hex_encode(
101            &self
102                .chain
103                .append(PLANE_CHAIN_SOURCE, kind, Some(payload))
104                .hash,
105        )
106    }
107
108    /// Place an operator-pinned store cog on a Seed. Pinned by
109    /// construction: the node is the operator-assigned id.
110    pub async fn place_store_pin(&self, o: &StorePinOrder) -> Result<PlacementRecord, PlaneError> {
111        let (host, tier) = self
112            .seed(&o.node_id)
113            .ok_or_else(|| PlaneError::Unknown(format!("no Seed adapter for {}", o.node_id)))?;
114        let w = VerifiedWorkload::store_pin(&o.registry, &o.id, &o.version, o.sha256.as_deref())
115            .map_err(|e| PlaneError::Invalid(e.to_string()))?;
116        let ctx = json!({ "node_id": o.node_id, "workload": {
117            "kind": w.kind, "package_trust": "operator_attested", "node_tier": governance_tier(tier),
118            "network": host.runtime().network_exposure(), "secrets": false, "emulated": false,
119            "resource_cost": 0.25, "package_id": format!("store.{}.{}", w.id, w.version),
120        }});
121        let verdict = self.gate.check(&self.node_id, "workload.place", &ctx);
122        let permitted = matches!(verdict, GateDecision::Permit { .. });
123        let decision_id = self.chain_here(
124            chain::EVENT_KIND_WORKLOAD_PLACE,
125            json!({ "phase": "decision", "route": SEED_ROUTE, "runtime": host.runtime().id(),
126                    "pin": o.node_id, "store_pin": { "registry": o.registry, "id": o.id,
127                    "version": o.version }, "permitted": permitted }),
128        );
129        if !permitted {
130            let reason = match verdict {
131                GateDecision::Deny { reason, .. } | GateDecision::Defer { reason } => reason,
132                GateDecision::Permit { .. } => String::new(),
133            };
134            return Err(PlaneError::Governance(reason));
135        }
136        let cfg = WorkloadConfig {
137            mode: o.config.mode.clone(),
138            args: o.config.args.clone(),
139            host: HostContract::new(SocketAddr::from(([0, 0, 0, 0], o.config.csi_port))),
140            node_id: o.node_id.clone(),
141        };
142        let refuse = |phase: &str, e: &dyn std::fmt::Display| {
143            self.chain_here(
144                chain::EVENT_KIND_WORKLOAD_REFUSE,
145                json!({ "phase": phase, "decision_id": decision_id, "node": o.node_id,
146                        "route": SEED_ROUTE, "reason": e.to_string() }),
147            );
148        };
149        let h = match host.load(&w, &cfg).await {
150            Ok(h) => h,
151            Err(e) => {
152                refuse("dispatch", &e);
153                return Err(PlaneError::Call(super::plane::CallFailure::Refused(
154                    super::msg::Refusal::new(super::msg::RefusalCode::Admission, e.to_string()),
155                )));
156            }
157        };
158        if o.start
159            && let Err(e) = host.start(&h).await
160        {
161            // Not placed: forget the loaded instance so nothing is left
162            // that the controller does not track.
163            let rolled = host.unload(h.clone()).await;
164            let why = match &rolled {
165                Ok(()) => format!("start failed: {e}; loaded instance unloaded"),
166                Err(u) => format!("start failed: {e}; unload failed too: {u}"),
167            };
168            refuse("start", &why);
169            if rolled.is_err() {
170                // Still loaded: keep it addressable (status / unload).
171                self.track_seed(self.seed_record(o, &w, &h, &decision_id), h, o);
172            }
173            return Err(PlaneError::Call(super::plane::CallFailure::Refused(
174                super::msg::Refusal::new(super::msg::RefusalCode::Runtime, why),
175            )));
176        }
177        let rec = self.seed_record(o, &w, &h, &decision_id);
178        self.chain_here(
179            chain::EVENT_KIND_WORKLOAD_PLACE,
180            json!({ "phase": "placed", "decision_id": decision_id, "node": o.node_id,
181                    "route": SEED_ROUTE, "instance_id": h.instance_id }),
182        );
183        self.track_seed(rec.clone(), h, o);
184        Ok(rec)
185    }
186
187    fn seed_record(
188        &self,
189        o: &StorePinOrder,
190        w: &VerifiedWorkload,
191        h: &InstanceHandle,
192        decision_id: &str,
193    ) -> PlacementRecord {
194        PlacementRecord {
195            instance_id: h.instance_id.clone(),
196            node_id: o.node_id.clone(),
197            kind: w.kind.clone(),
198            workload: w.id.clone(),
199            variant: SEED_ROUTE.to_string(),
200            decision_id: decision_id.to_string(),
201            manifest_hash: String::new(),
202            project_id: None,
203            package_id: Some(w.revocation_refs().0),
204            signer_keys: Vec::new(),
205            artifact_hashes: Vec::new(),
206        }
207    }
208
209    fn track_seed(&self, rec: PlacementRecord, h: InstanceHandle, o: &StorePinOrder) {
210        if let Ok(mut m) = self.seed_handles.lock() {
211            m.insert(
212                h.instance_id.clone(),
213                SeedEntry {
214                    handle: h,
215                    registry: o.registry.clone(),
216                    version: o.version.clone(),
217                    sha256: o.sha256.clone(),
218                },
219            );
220        }
221        if let Ok(mut p) = self.placements.lock() {
222            p.insert(rec.instance_id.clone(), rec);
223        }
224        self.persist();
225    }
226
227    /// Instance verbs on a Seed placement (gated and chained by its host).
228    pub(super) async fn seed_instance(
229        &self,
230        rec: &PlacementRecord,
231        m: &str,
232    ) -> Option<Result<Value, PlaneError>> {
233        let (host, _) = self.seed(&rec.node_id)?;
234        let entry = self
235            .seed_handles
236            .lock()
237            .ok()
238            .and_then(|m| m.get(&rec.instance_id).cloned());
239        let Some(SeedEntry {
240            handle: h,
241            registry,
242            version,
243            sha256,
244        }) = entry
245        else {
246            return Some(Err(PlaneError::Unknown(rec.instance_id.clone())));
247        };
248        let err = |e: crate::workload_runtime::RuntimeError| PlaneError::Governance(e.to_string());
249        // After a controller restart the adapter has no table: re-attach
250        // the instance (the Seed must still hold the pinned cog) so every
251        // verb below works on it again.
252        let adopted = async {
253            let w = VerifiedWorkload::store_pin(
254                &registry,
255                &h.workload_id,
256                &version,
257                sha256.as_deref(),
258            )?;
259            host.adopt(&h, &w).await
260        }
261        .await;
262        if let Err(e) = adopted {
263            // The Seed no longer holds the cog (removed out of band): an
264            // unload has nothing left to do but forget the record.
265            if m == method::UNLOAD && matches!(e, RuntimeError::NotInstalled(_)) {
266                if let Ok(mut s) = self.seed_handles.lock() {
267                    s.remove(&rec.instance_id);
268                }
269                if let Ok(mut p) = self.placements.lock() {
270                    p.remove(&rec.instance_id);
271                }
272                self.persist();
273                return Some(Ok(
274                    json!({ "unloaded": rec.instance_id, "note": "not on the Seed" }),
275                ));
276            }
277            return Some(Err(err(e)));
278        }
279        Some(match m {
280            method::STATUS => {
281                Ok(json!({ "instance_id": h.instance_id, "status": host.status(&h).await }))
282            }
283            method::START => host
284                .start(&h)
285                .await
286                .map(|_| json!({ "started": h.instance_id }))
287                .map_err(err),
288            method::STOP => host
289                .stop(&h, Duration::from_secs(5))
290                .await
291                .map(|ev| json!({ "stopped": h.instance_id, "evidence": ev.audit() }))
292                .map_err(err),
293            method::UNLOAD => {
294                let r = host.unload(h.clone()).await;
295                if r.is_ok() {
296                    if let Ok(mut m) = self.seed_handles.lock() {
297                        m.remove(&rec.instance_id);
298                    }
299                    if let Ok(mut p) = self.placements.lock() {
300                        p.remove(&rec.instance_id);
301                    }
302                    self.persist();
303                }
304                r.map(|_| json!({ "unloaded": rec.instance_id }))
305                    .map_err(err)
306            }
307            other => Err(PlaneError::Invalid(format!(
308                "{other} is not supported on a Seed"
309            ))),
310        })
311    }
312}
clawft_kernel/workload_ctl/plane_seed.rs

clawft_kernel/workload_ctl/
plane_seed.rs
