a3s_runtime/contract/
observation.rs

1use super::{
2    ArtifactRef, IsolationLevel, NetworkMode, RuntimeOutputArtifact, RuntimeServiceEndpoint,
3    RuntimeUnitClass, RuntimeUnitSpec,
4};
5use serde::{Deserialize, Serialize};
6use std::collections::{BTreeMap, BTreeSet};
7
8#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
9#[serde(rename_all = "snake_case")]
10pub enum RuntimeUnitState {
11    Accepted,
12    Preparing,
13    Starting,
14    Running,
15    Stopping,
16    Stopped,
17    Succeeded,
18    Failed,
19    Unknown,
20}
21
22impl RuntimeUnitState {
23    pub fn is_terminal(self) -> bool {
24        matches!(self, Self::Stopped | Self::Succeeded | Self::Failed)
25    }
26}
27
28#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
29#[serde(rename_all = "snake_case")]
30pub enum RuntimeHealthState {
31    Unknown,
32    Starting,
33    Healthy,
34    Unhealthy,
35}
36
37#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
38#[serde(deny_unknown_fields)]
39pub struct RuntimeHealthObservation {
40    pub state: RuntimeHealthState,
41    pub checked_at_ms: u64,
42    pub message: Option<String>,
43}
44
45#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
46#[serde(deny_unknown_fields)]
47pub struct RuntimeUsage {
48    pub wall_time_ms: u64,
49    pub cpu_time_ms: u64,
50    pub peak_memory_bytes: u64,
51    pub network_rx_bytes: u64,
52    pub network_tx_bytes: u64,
53    pub storage_read_bytes: u64,
54    pub storage_write_bytes: u64,
55}
56
57#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
58#[serde(deny_unknown_fields)]
59pub struct RuntimeFailure {
60    pub code: String,
61    pub message: String,
62    pub retryable: bool,
63}
64
65impl RuntimeFailure {
66    fn validate(&self) -> Result<(), String> {
67        super::validate_name("failure code", &self.code)?;
68        super::validate_nonempty("failure message", &self.message, 16 * 1024)
69    }
70}
71
72#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
73#[serde(deny_unknown_fields)]
74pub struct RuntimeEvidence {
75    pub provider_build: String,
76    pub spec_digest: String,
77    pub semantics_profile_digest: Option<String>,
78    pub identity_attachment_digest: Option<String>,
79    pub claims: BTreeMap<String, String>,
80}
81
82impl RuntimeEvidence {
83    pub(crate) fn validate(&self) -> Result<(), String> {
84        super::validate_nonempty("provider_build", &self.provider_build, 255)?;
85        super::validate_digest(&self.spec_digest)?;
86        if let Some(digest) = &self.semantics_profile_digest {
87            super::validate_digest(digest)?;
88        }
89        if let Some(digest) = &self.identity_attachment_digest {
90            super::validate_digest(digest)?;
91        }
92        if self.claims.len() > 128
93            || self
94                .claims
95                .iter()
96                .any(|(key, value)| key.len() > 255 || value.len() > 4096)
97        {
98            return Err("Runtime evidence claims exceed protocol limits".into());
99        }
100        Ok(())
101    }
102}
103
104#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
105#[serde(deny_unknown_fields)]
106pub struct RuntimeObservation {
107    pub schema: String,
108    pub unit_id: String,
109    pub generation: u64,
110    pub spec_digest: String,
111    pub class: RuntimeUnitClass,
112    pub state: RuntimeUnitState,
113    pub provider_resource_id: Option<String>,
114    pub provider_build: Option<String>,
115    pub observed_at_ms: u64,
116    pub started_at_ms: Option<u64>,
117    pub finished_at_ms: Option<u64>,
118    /// Readiness observation corresponding to `RuntimeUnitSpec::health`.
119    pub health: Option<RuntimeHealthObservation>,
120    /// Liveness observation corresponding to the optional Service lifecycle.
121    pub liveness: Option<RuntimeHealthObservation>,
122    pub outputs: Vec<RuntimeOutputArtifact>,
123    pub usage: Option<RuntimeUsage>,
124    pub evidence: Option<RuntimeEvidence>,
125    pub provider_attestation: Option<ArtifactRef>,
126    pub failure: Option<RuntimeFailure>,
127}
128
129impl RuntimeObservation {
130    pub const SCHEMA: &'static str = "a3s.runtime.observation.v4";
131
132    pub(crate) fn accepted(spec: &RuntimeUnitSpec, observed_at_ms: u64) -> Result<Self, String> {
133        Ok(Self {
134            schema: Self::SCHEMA.into(),
135            unit_id: spec.unit_id.clone(),
136            generation: spec.generation,
137            spec_digest: spec.digest()?,
138            class: spec.class,
139            state: RuntimeUnitState::Accepted,
140            provider_resource_id: None,
141            provider_build: None,
142            observed_at_ms,
143            started_at_ms: None,
144            finished_at_ms: None,
145            health: None,
146            liveness: None,
147            outputs: Vec::new(),
148            usage: None,
149            evidence: None,
150            provider_attestation: None,
151            failure: None,
152        })
153    }
154
155    pub fn validate(&self) -> Result<(), String> {
156        if self.schema != Self::SCHEMA {
157            return Err(format!(
158                "unsupported Runtime observation schema {:?}",
159                self.schema
160            ));
161        }
162        super::validate_id("unit_id", &self.unit_id, 512)?;
163        if self.generation == 0 {
164            return Err("Runtime observation generation must be positive".into());
165        }
166        super::validate_digest(&self.spec_digest)?;
167        if let Some(value) = &self.provider_resource_id {
168            super::validate_nonempty("provider_resource_id", value, 1024)?;
169        }
170        if let Some(value) = &self.provider_build {
171            super::validate_nonempty("provider_build", value, 255)?;
172        }
173        if !matches!(
174            self.state,
175            RuntimeUnitState::Accepted | RuntimeUnitState::Unknown
176        ) && (self.provider_resource_id.is_none() || self.provider_build.is_none())
177        {
178            return Err("provider-backed observations require resource and build identity".into());
179        }
180        if let (Some(started), Some(finished)) = (self.started_at_ms, self.finished_at_ms) {
181            if finished < started {
182                return Err("finished_at_ms precedes started_at_ms".into());
183            }
184        }
185        if self.state.is_terminal() != self.finished_at_ms.is_some() {
186            return Err("terminal state and finished_at_ms do not agree".into());
187        }
188        if self.state == RuntimeUnitState::Failed {
189            self.failure
190                .as_ref()
191                .ok_or_else(|| "failed observation is missing failure".to_string())?
192                .validate()?;
193        } else if self.failure.is_some() {
194            return Err("non-failed observation contains failure".into());
195        }
196        if self.class == RuntimeUnitClass::Service && self.state == RuntimeUnitState::Succeeded {
197            return Err("Service cannot enter succeeded state".into());
198        }
199        if self.class == RuntimeUnitClass::Task
200            && (self.health.is_some() || self.liveness.is_some())
201        {
202            return Err("Task observation cannot contain Service readiness or liveness".into());
203        }
204        if !(self.outputs.is_empty()
205            || self.class == RuntimeUnitClass::Task && self.state == RuntimeUnitState::Succeeded)
206        {
207            return Err("output artifacts require a succeeded Task".into());
208        }
209        let mut output_names = BTreeSet::new();
210        for output in &self.outputs {
211            output.validate()?;
212            if !output_names.insert(&output.name) {
213                return Err(format!("duplicate output artifact {:?}", output.name));
214            }
215        }
216        if let Some(health) = &self.health {
217            if let Some(message) = &health.message {
218                super::validate_nonempty("health message", message, 4096)?;
219            }
220        }
221        if let Some(liveness) = &self.liveness {
222            if let Some(message) = &liveness.message {
223                super::validate_nonempty("liveness message", message, 4096)?;
224            }
225        }
226        if let Some(evidence) = &self.evidence {
227            evidence.validate()?;
228            if evidence.spec_digest != self.spec_digest {
229                return Err("Runtime evidence does not bind the observation spec".into());
230            }
231            if self.provider_build.as_ref() != Some(&evidence.provider_build) {
232                return Err("Runtime evidence does not bind the observation provider build".into());
233            }
234        }
235        let endpoints = self.service_endpoints()?;
236        if !endpoints.is_empty()
237            && (self.class != RuntimeUnitClass::Service || self.state != RuntimeUnitState::Running)
238        {
239            return Err("Runtime service endpoints require a running Service observation".into());
240        }
241        let mut endpoint_sockets = BTreeSet::new();
242        if endpoints
243            .iter()
244            .any(|endpoint| !endpoint_sockets.insert((endpoint.protocol, endpoint.socket_addr())))
245        {
246            return Err("Runtime observation contains duplicate service endpoint sockets".into());
247        }
248        if let Some(attestation) = &self.provider_attestation {
249            attestation.validate()?;
250            if self.provider_resource_id.is_none()
251                || self.provider_build.is_none()
252                || self.evidence.is_none()
253            {
254                return Err(
255                    "Runtime provider attestation requires resource, build, and evidence identity"
256                        .into(),
257                );
258            }
259        }
260        Ok(())
261    }
262
263    pub fn validate_against(&self, spec: &RuntimeUnitSpec) -> Result<(), String> {
264        self.validate()?;
265        spec.validate()?;
266        if self.unit_id != spec.unit_id
267            || self.generation != spec.generation
268            || self.class != spec.class
269            || self.spec_digest != spec.digest()?
270        {
271            return Err("Runtime observation does not match the unit specification".into());
272        }
273        if self.evidence.as_ref().is_some_and(|evidence| {
274            evidence.semantics_profile_digest != spec.semantics_profile_digest
275        }) {
276            return Err(
277                "Runtime evidence semantics profile does not match the unit specification".into(),
278            );
279        }
280        if self.provider_resource_id.is_some()
281            && self
282                .evidence
283                .as_ref()
284                .and_then(|evidence| evidence.identity_attachment_digest.as_ref())
285                != spec.identity_attachment_digest.as_ref()
286        {
287            return Err(
288                "Runtime evidence identity attachment does not match the unit specification".into(),
289            );
290        }
291        if self.state == RuntimeUnitState::Succeeded {
292            if self.outputs.len() != spec.outputs.len() {
293                return Err("succeeded Task did not report the exact requested outputs".into());
294            }
295            for expected in &spec.outputs {
296                let output = self
297                    .outputs
298                    .iter()
299                    .find(|output| output.name == expected.name)
300                    .ok_or_else(|| format!("succeeded Task omitted output {:?}", expected.name))?;
301                if output.artifact.media_type != expected.media_type {
302                    return Err(format!(
303                        "output {:?} media type does not match its specification",
304                        expected.name
305                    ));
306                }
307                if output.size_bytes > expected.max_bytes {
308                    return Err(format!(
309                        "output {:?} exceeds its maximum size",
310                        expected.name
311                    ));
312                }
313            }
314        } else if !self.outputs.is_empty() {
315            return Err("only a succeeded Task may report outputs".into());
316        }
317        let running_service =
318            self.class == RuntimeUnitClass::Service && self.state == RuntimeUnitState::Running;
319        if running_service {
320            if self.health.is_some() != spec.health.is_some() {
321                return Err(
322                    "Runtime readiness observation does not match the Service specification".into(),
323                );
324            }
325            if self.liveness.is_some() != spec.service_lifecycle.is_some() {
326                return Err(
327                    "Runtime liveness observation does not match the Service lifecycle".into(),
328                );
329            }
330        } else if self.health.is_some() || self.liveness.is_some() {
331            return Err("only a running Service may report readiness or liveness".into());
332        }
333        let endpoints = self.service_endpoints()?;
334        if self.class == RuntimeUnitClass::Service
335            && self.state == RuntimeUnitState::Running
336            && spec.network.mode == NetworkMode::Service
337        {
338            if endpoints.len() != spec.network.ports.len() {
339                return Err(
340                    "running Runtime Service did not report the exact declared endpoints".into(),
341                );
342            }
343            for port in &spec.network.ports {
344                let endpoint = endpoints
345                    .iter()
346                    .find(|endpoint| endpoint.port_name == port.name)
347                    .ok_or_else(|| {
348                        format!("running Runtime Service omitted endpoint {:?}", port.name)
349                    })?;
350                if endpoint.protocol != port.protocol {
351                    return Err(format!(
352                        "Runtime service endpoint {:?} protocol does not match its declaration",
353                        port.name
354                    ));
355                }
356            }
357        } else if !endpoints.is_empty() {
358            return Err("Runtime service endpoints do not match the unit lifecycle".into());
359        }
360        if spec.isolation == IsolationLevel::Confidential
361            && self.provider_resource_id.is_some()
362            && self.provider_attestation.is_none()
363        {
364            return Err(
365                "provider-backed confidential Runtime observation requires attestation".into(),
366            );
367        }
368        Ok(())
369    }
370
371    pub fn service_endpoints(&self) -> Result<Vec<RuntimeServiceEndpoint>, String> {
372        self.evidence
373            .as_ref()
374            .map(|evidence| RuntimeServiceEndpoint::from_claims(&evidence.claims))
375            .unwrap_or_else(|| Ok(Vec::new()))
376    }
377
378    pub fn clear_service_endpoints(&mut self) {
379        if let Some(evidence) = &mut self.evidence {
380            RuntimeServiceEndpoint::remove_claims(&mut evidence.claims);
381        }
382    }
383
384    pub fn converges(&self, spec: &RuntimeUnitSpec) -> bool {
385        if self.validate_against(spec).is_err() {
386            return false;
387        }
388        match spec.class {
389            RuntimeUnitClass::Task => self.state == RuntimeUnitState::Succeeded,
390            RuntimeUnitClass::Service => {
391                self.state == RuntimeUnitState::Running
392                    && spec.health.as_ref().is_none_or(|_| {
393                        self.health
394                            .as_ref()
395                            .is_some_and(|health| health.state == RuntimeHealthState::Healthy)
396                    })
397                    && spec.service_lifecycle.as_ref().is_none_or(|_| {
398                        self.liveness
399                            .as_ref()
400                            .is_some_and(|health| health.state == RuntimeHealthState::Healthy)
401                    })
402            }
403        }
404    }
405}
406
407#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
408#[serde(tag = "status", rename_all = "snake_case", deny_unknown_fields)]
409pub enum RuntimeInspection {
410    Found {
411        schema: String,
412        observation: Box<RuntimeObservation>,
413    },
414    NotFound {
415        schema: String,
416        unit_id: String,
417        last_generation: Option<u64>,
418    },
419}
420
421impl RuntimeInspection {
422    pub const SCHEMA: &'static str = "a3s.runtime.inspection.v1";
423
424    pub fn validate(&self) -> Result<(), String> {
425        match self {
426            Self::Found {
427                schema,
428                observation,
429            } => {
430                validate_inspection_schema(schema)?;
431                observation.validate()
432            }
433            Self::NotFound {
434                schema,
435                unit_id,
436                last_generation,
437            } => {
438                validate_inspection_schema(schema)?;
439                super::validate_id("unit_id", unit_id, 512)?;
440                if *last_generation == Some(0) {
441                    return Err("last_generation must be positive when present".into());
442                }
443                Ok(())
444            }
445        }
446    }
447}
448
449fn validate_inspection_schema(schema: &str) -> Result<(), String> {
450    if schema != RuntimeInspection::SCHEMA {
451        return Err(format!("unsupported Runtime inspection schema {schema:?}"));
452    }
453    Ok(())
454}
a3s_runtime/contract/observation.rs

a3s_runtime/contract/
observation.rs
