clawft_kernel/workload_runtime/
host.rs

1//! [`WorkloadHost`]: gates every adapter transition through governance and
2//! chains what the adapter did (ADR-099 sections 4 and 5).
3//!
4//! The gate chains its own decision under the action's kind with source
5//! `workload`; the host then chains the transition's outcome under the same
6//! kind with source [`RUNTIME_CHAIN_SOURCE`]. Admission refusals and adapter
7//! failures are chained as `workload.refuse`.
8
9use std::collections::HashMap;
10use std::sync::Arc;
11use std::time::Duration;
12
13use serde_json::{Value, json};
14use tokio::sync::Mutex;
15
16use super::cog_spec::valid_value;
17use super::evidence::RunEvidence;
18use super::types::{
19    ControlMode, InstanceHandle, InstanceStatus, Preemption, RuntimeError, VerifiedWorkload, WorkloadConfig,
20    WorkloadRuntime, WorkloadSource,
21};
22use crate::chain::{self, ChainManager};
23use crate::gate::{GateBackend, GateDecision};
24use crate::workload_governance::{NetworkPolicy, NodeTrustTier};
25
26/// Chain source for adapter lifecycle outcomes.
27pub const RUNTIME_CHAIN_SOURCE: &str = "workload.runtime";
28
29/// Governs and records one adapter.
30pub struct WorkloadHost {
31    runtime: Arc<dyn WorkloadRuntime>,
32    gate: Arc<dyn GateBackend>,
33    chain: Option<Arc<ChainManager>>,
34    agent_id: String,
35    /// Live: the control plane lowers it when the operator demotes the node.
36    node_tier: std::sync::RwLock<NodeTrustTier>,
37    loaded: Mutex<HashMap<String, (VerifiedWorkload, bool)>>,
38}
39
40impl WorkloadHost {
41    /// Host for `runtime` on a node of `node_tier`.
42    pub fn new(
43        runtime: Arc<dyn WorkloadRuntime>,
44        gate: Arc<dyn GateBackend>,
45        agent_id: impl Into<String>,
46        node_tier: NodeTrustTier,
47    ) -> Self {
48        Self {
49            runtime,
50            gate,
51            chain: None,
52            agent_id: agent_id.into(),
53            node_tier: std::sync::RwLock::new(node_tier),
54            loaded: Mutex::new(HashMap::new()),
55        }
56    }
57
58    /// Attach the chain.
59    pub fn with_chain(mut self, cm: Arc<ChainManager>) -> Self {
60        self.chain = Some(cm);
61        self
62    }
63
64    /// The trust tier this host's node is governed at right now.
65    pub fn node_tier(&self) -> NodeTrustTier {
66        self.node_tier
67            .read()
68            .map_or(NodeTrustTier::Discovered, |t| *t)
69    }
70
71    /// Change the governed tier (an operator demotion or promotion). Takes
72    /// effect on the next gated call, including on instances already placed.
73    pub fn set_node_tier(&self, tier: NodeTrustTier) {
74        if let Ok(mut t) = self.node_tier.write() {
75            *t = tier;
76        }
77    }
78
79    /// The adapter.
80    pub fn runtime(&self) -> &Arc<dyn WorkloadRuntime> {
81        &self.runtime
82    }
83
84    /// Governance context. `network` is what the adapter actually gives
85    /// instances ([`WorkloadRuntime::network_exposure`]). `secrets` is
86    /// false because no adapter delivers operator secrets: the only
87    /// credential an instance gets is its own `COGNITUM_COG_TOKEN`, minted
88    /// here for that instance's ingest bridge (ADR-100 section 4).
89    fn context(&self, w: &VerifiedWorkload, kind: &str, emulated: bool) -> Value {
90        let network: NetworkPolicy = self.runtime.network_exposure();
91        let (package_id, keys, hashes) = w.revocation_refs();
92        let (trust, cost) = match &w.source {
93            WorkloadSource::SignedPackage(p) => (
94                "pinned_signer",
95                (f64::from(p.spec.resources.cpu_pct) / 400.0).clamp(0.0, 1.0),
96            ),
97            WorkloadSource::StorePin { .. } => ("operator_attested", 0.25),
98            // `PackageTrust::ProjectCert`: matched only by a permit that
99            // names it (the supervisor's, see `project_supervisor_permit`).
100            WorkloadSource::Project(_) => ("project_cert", 0.25),
101            // Managed: model weights are operator-attested hash manifests
102            // (ADR-101 section 3). Adopted: a server the operator started
103            // is only observed, nothing is verified. Cost is the share of a
104            // 128 GiB unified pool.
105            WorkloadSource::Inference(p) => {
106                let m = &p.spec.memory;
107                let bytes = m.weights_bytes.saturating_add(m.kv_budget_bytes);
108                let trust = if self.runtime.control_mode() == ControlMode::Adopted {
109                    "adopted_unverified"
110                } else {
111                    "operator_attested"
112                };
113                (
114                    trust,
115                    (bytes as f64 / (128.0 * 1024.0 * 1024.0 * 1024.0)).clamp(0.1, 1.0),
116                )
117            }
118        };
119        json!({ "workload": {
120            "kind": kind,
121            "package_trust": trust,
122            "node_tier": self.node_tier(),
123            "network": network,
124            "secrets": false,
125            "emulated": emulated,
126            "resource_cost": cost,
127            "package_id": package_id,
128            "signer_keys": keys,
129            "artifact_hashes": hashes,
130        }})
131    }
132
133    pub(super) fn check(
134        &self,
135        action: &str,
136        w: &VerifiedWorkload,
137        kind: &str,
138        emulated: bool,
139    ) -> Result<(), RuntimeError> {
140        self.verdict(action, w, kind, emulated, false)
141    }
142
143    /// [`Self::check`] for taking down what is already placed (stop,
144    /// unload, re-adoption): a denial caused by the node tier alone is
145    /// waived by the gate ([`GateBackend::check_teardown`]).
146    fn check_teardown(
147        &self,
148        action: &str,
149        w: &VerifiedWorkload,
150        kind: &str,
151        emulated: bool,
152    ) -> Result<(), RuntimeError> {
153        self.verdict(action, w, kind, emulated, true)
154    }
155
156    fn verdict(
157        &self,
158        action: &str,
159        w: &VerifiedWorkload,
160        kind: &str,
161        emulated: bool,
162        teardown: bool,
163    ) -> Result<(), RuntimeError> {
164        let ctx = self.context(w, kind, emulated);
165        let d = if teardown {
166            self.gate.check_teardown(&self.agent_id, action, &ctx)
167        } else {
168            self.gate.check(&self.agent_id, action, &ctx)
169        };
170        match d {
171            GateDecision::Permit { .. } => Ok(()),
172            GateDecision::Deny { reason, .. } => Err(RuntimeError::Governance(format!(
173                "{action} denied: {reason}"
174            ))),
175            GateDecision::Defer { reason } => Err(RuntimeError::Governance(format!(
176                "{action} deferred to a human: {reason}"
177            ))),
178        }
179    }
180
181    fn record(&self, kind: &str, payload: Value) {
182        if let Some(cm) = &self.chain {
183            cm.append(RUNTIME_CHAIN_SOURCE, kind, Some(payload));
184        }
185    }
186
187    fn base(&self, w: &VerifiedWorkload, instance: Option<&str>) -> Value {
188        json!({
189            "runtime": self.runtime.id(),
190            "workload_id": w.id,
191            "version": w.version,
192            "instance_id": instance,
193        })
194    }
195
196    pub(super) fn outcome<T>(&self, kind: &str, mut payload: Value, r: &Result<T, RuntimeError>) {
197        match r {
198            Ok(_) => {
199                payload["outcome"] = json!("ok");
200                self.record(kind, payload);
201            }
202            Err(e) => {
203                payload["outcome"] = json!("error");
204                payload["failed_action"] = json!(kind);
205                payload["error_code"] = json!(e.code());
206                payload["reason"] = json!(e.to_string());
207                self.record(chain::EVENT_KIND_WORKLOAD_REFUSE, payload);
208            }
209        }
210    }
211
212    async fn workload_of(
213        &self,
214        h: &InstanceHandle,
215    ) -> Result<(VerifiedWorkload, bool), RuntimeError> {
216        self.loaded
217            .lock()
218            .await
219            .get(&h.instance_id)
220            .cloned()
221            .ok_or_else(|| RuntimeError::UnknownInstance(h.instance_id.clone()))
222    }
223
224    /// Gate, admit and load. Store pins are also gated as `workload.install`.
225    pub async fn load(
226        &self,
227        w: &VerifiedWorkload,
228        cfg: &WorkloadConfig,
229    ) -> Result<InstanceHandle, RuntimeError> {
230        let admission = self.runtime.admit(w).await;
231        let emulated = admission.as_ref().is_ok_and(|a| a.emulated);
232        let mut payload = self.base(w, None);
233        payload["admission"] = json!(admission.as_ref().ok());
234        payload["host_contract"] = cfg.host.audit();
235        if let Err(e) = &admission {
236            self.outcome::<()>(chain::EVENT_KIND_WORKLOAD_LOAD, payload, &Err(e.clone()));
237            return Err(e.clone());
238        }
239        let store = matches!(w.source, WorkloadSource::StorePin { .. });
240        if store {
241            self.check("workload.install", w, &w.kind, emulated)?;
242        }
243        self.check("workload.load", w, &w.kind, emulated)?;
244        let r = self.runtime.load(w, cfg).await;
245        if let Ok(h) = &r {
246            payload["instance_id"] = json!(h.instance_id);
247            self.loaded
248                .lock()
249                .await
250                .insert(h.instance_id.clone(), (w.clone(), emulated));
251            // Chained only when something was actually installed.
252            if h.store_installed {
253                self.record(
254                    chain::EVENT_KIND_WORKLOAD_INSTALL,
255                    json!({
256                        "runtime": self.runtime.id(), "workload_id": w.id, "version": w.version,
257                        "instance_id": h.instance_id, "outcome": "ok", "source": "store_pin",
258                    }),
259                );
260            }
261        }
262        if let Err(RuntimeError::StrandedInstall {
263            handle,
264            rolled_back,
265            ..
266        }) = &r
267        {
268            // The device store changed even though the load failed.
269            self.record(
270                chain::EVENT_KIND_WORKLOAD_INSTALL,
271                json!({
272                    "runtime": self.runtime.id(), "workload_id": w.id, "version": w.version,
273                    "instance_id": handle.instance_id, "source": "store_pin",
274                    "outcome": if *rolled_back { "rolled-back" } else { "stranded" },
275                }),
276            );
277            if !rolled_back {
278                // Still on the device: keep it unloadable through the host.
279                self.loaded
280                    .lock()
281                    .await
282                    .insert(handle.instance_id.clone(), (w.clone(), emulated));
283            }
284        }
285        self.outcome(chain::EVENT_KIND_WORKLOAD_LOAD, payload, &r);
286        r
287    }
288
289    /// Re-attach an instance loaded before a controller restart
290    /// ([`WorkloadRuntime::adopt`]), gated as `workload.load` and chained
291    /// with phase `readopted`. A no-op if the instance is already known.
292    pub async fn adopt(
293        &self,
294        h: &InstanceHandle,
295        w: &VerifiedWorkload,
296    ) -> Result<(), RuntimeError> {
297        if self.loaded.lock().await.contains_key(&h.instance_id) {
298            return Ok(());
299        }
300        self.check_teardown("workload.load", w, &w.kind, false)?;
301        let r = self.runtime.adopt(h, w).await;
302        if r.is_ok() {
303            self.loaded
304                .lock()
305                .await
306                .insert(h.instance_id.clone(), (w.clone(), false));
307        }
308        let mut payload = self.base(w, Some(&h.instance_id));
309        payload["phase"] = json!("readopted");
310        self.outcome(chain::EVENT_KIND_WORKLOAD_LOAD, payload, &r);
311        r
312    }
313
314    /// Gate and start.
315    pub async fn start(&self, h: &InstanceHandle) -> Result<(), RuntimeError> {
316        let (w, emu) = self.workload_of(h).await?;
317        self.check("workload.start", &w, &w.kind, emu)?;
318        let r = self.runtime.start(h).await;
319        self.outcome(
320            chain::EVENT_KIND_WORKLOAD_START,
321            self.base(&w, Some(&h.instance_id)),
322            &r,
323        );
324        r
325    }
326
327    /// Gate and run one console cycle, chained as `workload.start` with
328    /// phase `console`.
329    ///
330    /// Instances the adapter says must stop first (sensor feed contention,
331    /// including ones WeftOS did not load) are each gated as
332    /// `workload.stop`; if any is denied nothing is stopped. Each stop is
333    /// chained as it happens (so a failed console run still leaves the
334    /// record), and afterwards each stopped instance is restarted, gated
335    /// as `workload.start` and chained with phase `resume-after-console`.
336    pub async fn console(
337        &self,
338        h: &InstanceHandle,
339        command: &str,
340    ) -> Result<RunEvidence, RuntimeError> {
341        let (w, emu) = self.workload_of(h).await?;
342        self.check("workload.start", &w, &w.kind, emu)?;
343        let mut payload = self.base(&w, Some(&h.instance_id));
344        payload["phase"] = json!("console");
345        let pre = match self.preemptions_permitted(h).await {
346            Ok(p) => p,
347            Err(e) => {
348                self.outcome::<()>(chain::EVENT_KIND_WORKLOAD_START, payload, &Err(e.clone()));
349                return Err(e);
350            }
351        };
352        let mut stopped = Vec::new();
353        let mut failed = None;
354        for (p, _) in &pre {
355            let r = self.runtime.preempt(p).await;
356            self.outcome(
357                chain::EVENT_KIND_WORKLOAD_STOP,
358                self.preempt_payload(p, h, "preempt-for-console"),
359                &r,
360            );
361            match r {
362                Ok(()) => stopped.push(p),
363                Err(e) => {
364                    failed = Some(e);
365                    break;
366                }
367            }
368        }
369        let mut r = match failed {
370            Some(e) => Err(e),
371            None => self.runtime.console(h, command).await,
372        };
373        let mut resumed = Vec::new();
374        for (p, pw) in pre.iter().filter(|(p, _)| stopped.contains(&p)) {
375            if self.check("workload.start", pw, &pw.kind, false).is_err() {
376                continue; // the gate chained its denial; the cog stays stopped
377            }
378            let rr = self.runtime.resume(p).await;
379            self.outcome(
380                chain::EVENT_KIND_WORKLOAD_START,
381                self.preempt_payload(p, h, "resume-after-console"),
382                &rr,
383            );
384            if rr.is_ok() {
385                resumed.push(p.workload_id.clone());
386            }
387        }
388        let ids: Vec<String> = stopped.iter().map(|p| p.workload_id.clone()).collect();
389        payload["preempted"] = json!(ids);
390        payload["resumed"] = json!(resumed);
391        if let Ok(ev) = &mut r {
392            ev.stopped_for_console = ids;
393            payload["evidence"] = ev.audit();
394        }
395        self.outcome(chain::EVENT_KIND_WORKLOAD_START, payload, &r);
396        r
397    }
398
399    /// The adapter's console preemptions, each gated as `workload.stop`.
400    async fn preemptions_permitted(
401        &self,
402        h: &InstanceHandle,
403    ) -> Result<Vec<(Preemption, VerifiedWorkload)>, RuntimeError> {
404        let mut out = Vec::new();
405        for p in self.runtime.console_preemptions(h).await? {
406            let version = if valid_value(&p.version) {
407                p.version.as_str()
408            } else {
409                "unknown"
410            };
411            let pw = VerifiedWorkload::store_pin(&p.registry, &p.workload_id, version, None)?;
412            self.check("workload.stop", &pw, &pw.kind, false)
413                .map_err(|e| {
414                    RuntimeError::Governance(format!(
415                        "console run needs {} stopped: {e}",
416                        p.workload_id
417                    ))
418                })?;
419            out.push((p, pw));
420        }
421        Ok(out)
422    }
423
424    fn preempt_payload(&self, p: &Preemption, h: &InstanceHandle, phase: &str) -> Value {
425        json!({
426            "runtime": self.runtime.id(),
427            "workload_id": p.workload_id,
428            "version": p.version,
429            "phase": phase,
430            "for_instance": h.instance_id,
431        })
432    }
433
434    /// Gate and stop.
435    pub async fn stop(
436        &self,
437        h: &InstanceHandle,
438        grace: Duration,
439    ) -> Result<RunEvidence, RuntimeError> {
440        let (w, emu) = self.workload_of(h).await?;
441        self.check_teardown("workload.stop", &w, &w.kind, emu)?;
442        let r = self.runtime.stop(h, grace).await;
443        let mut payload = self.base(&w, Some(&h.instance_id));
444        if let Ok(ev) = &r {
445            payload["evidence"] = ev.audit();
446        }
447        self.outcome(chain::EVENT_KIND_WORKLOAD_STOP, payload, &r);
448        r
449    }
450
451    /// Gate and unload.
452    pub async fn unload(&self, h: InstanceHandle) -> Result<(), RuntimeError> {
453        let (w, emu) = self.workload_of(&h).await?;
454        self.check_teardown("workload.unload", &w, &w.kind, emu)?;
455        let payload = self.base(&w, Some(&h.instance_id));
456        let iid = h.instance_id.clone();
457        let r = self.runtime.unload(h).await;
458        if r.is_ok() {
459            self.loaded.lock().await.remove(&iid);
460        }
461        self.outcome(chain::EVENT_KIND_WORKLOAD_UNLOAD, payload, &r);
462        r
463    }
464
465    /// The workload a loaded instance came from, if this host loaded or
466    /// adopted it.
467    pub async fn workload_for(&self, h: &InstanceHandle) -> Option<VerifiedWorkload> {
468        self.loaded
469            .lock()
470            .await
471            .get(&h.instance_id)
472            .map(|(w, _)| w.clone())
473    }
474
475    /// Stop and unload an instance because its package, signer or artifact
476    /// was revoked. Not gated: the applied revocation is the authority
477    /// (`why` names it in both records), so enforcement cannot depend on an
478    /// operator having written a stop or unload permit. The unload is
479    /// attempted even when the stop fails (the instance may already have
480    /// exited). Each step is chained like the gated one, with
481    /// `forced_by_revocation` set.
482    pub async fn revoke_teardown(
483        &self,
484        h: &InstanceHandle,
485        grace: Duration,
486        why: Value,
487    ) -> Result<(), RuntimeError> {
488        let (w, _emu) = self.workload_of(h).await?;
489        let mut payload = self.base(&w, Some(&h.instance_id));
490        payload["forced_by_revocation"] = why;
491        let stopped = self.runtime.stop(h, grace).await;
492        let mut stop_payload = payload.clone();
493        if let Ok(ev) = &stopped {
494            stop_payload["evidence"] = ev.audit();
495        }
496        self.outcome(chain::EVENT_KIND_WORKLOAD_STOP, stop_payload, &stopped);
497        let unloaded = self.runtime.unload(h.clone()).await;
498        if unloaded.is_ok() {
499            self.loaded.lock().await.remove(&h.instance_id);
500        }
501        self.outcome(chain::EVENT_KIND_WORKLOAD_UNLOAD, payload, &unloaded);
502        unloaded
503    }
504
505    /// Status (not gated: read-only).
506    pub async fn status(&self, h: &InstanceHandle) -> InstanceStatus {
507        self.runtime.status(h).await
508    }
509}
clawft_kernel/workload_runtime/host.rs

clawft_kernel/workload_runtime/
host.rs
