a3s_runtime/contract/
unit.rs

1use super::{
2    ArtifactRef, IsolationLevel, ResourceLimits, RuntimeNetworkSpec, RuntimeProcessSpec,
3    SecretReference,
4};
5use serde::{Deserialize, Serialize};
6use sha2::{Digest, Sha256};
7use std::collections::BTreeSet;
8
9#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
10#[serde(rename_all = "snake_case")]
11pub enum RuntimeUnitClass {
12    Task,
13    Service,
14}
15
16#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
17#[serde(rename_all = "snake_case")]
18pub enum MountKind {
19    Artifact,
20    Volume,
21    Tmpfs,
22}
23
24#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
25#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
26pub enum RuntimeMountSource {
27    Artifact { artifact: ArtifactRef },
28    Volume { volume_id: String },
29    Tmpfs { size_bytes: u64 },
30}
31
32impl RuntimeMountSource {
33    pub fn kind(&self) -> MountKind {
34        match self {
35            Self::Artifact { .. } => MountKind::Artifact,
36            Self::Volume { .. } => MountKind::Volume,
37            Self::Tmpfs { .. } => MountKind::Tmpfs,
38        }
39    }
40
41    fn validate(&self) -> Result<(), String> {
42        match self {
43            Self::Artifact { artifact } => artifact.validate(),
44            Self::Volume { volume_id } => super::validate_id("volume_id", volume_id, 255),
45            Self::Tmpfs { size_bytes } if *size_bytes == 0 => {
46                Err("tmpfs size_bytes must be positive".into())
47            }
48            Self::Tmpfs { .. } => Ok(()),
49        }
50    }
51}
52
53#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
54#[serde(deny_unknown_fields)]
55pub struct RuntimeMount {
56    pub name: String,
57    pub source: RuntimeMountSource,
58    pub target: String,
59    pub read_only: bool,
60}
61
62impl RuntimeMount {
63    fn validate(&self) -> Result<(), String> {
64        super::validate_name("mount name", &self.name)?;
65        super::validate_absolute_path("mount target", &self.target)?;
66        self.source.validate()
67    }
68}
69
70#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
71#[serde(rename_all = "snake_case")]
72pub enum HealthCheckKind {
73    Http,
74    Tcp,
75    Command,
76}
77
78#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
79#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
80pub enum HealthProbe {
81    Http {
82        port: String,
83        path: String,
84        expected_statuses: Vec<u16>,
85    },
86    Tcp {
87        port: String,
88    },
89    Command {
90        command: Vec<String>,
91    },
92}
93
94impl HealthProbe {
95    pub fn kind(&self) -> HealthCheckKind {
96        match self {
97            Self::Http { .. } => HealthCheckKind::Http,
98            Self::Tcp { .. } => HealthCheckKind::Tcp,
99            Self::Command { .. } => HealthCheckKind::Command,
100        }
101    }
102
103    fn validate(&self, network: &RuntimeNetworkSpec) -> Result<(), String> {
104        match self {
105            Self::Http {
106                port,
107                path,
108                expected_statuses,
109            } => {
110                super::validate_name("health port", port)?;
111                if !network.has_port(port) {
112                    return Err(format!(
113                        "HTTP health check references unknown port {port:?}"
114                    ));
115                }
116                if !path.starts_with('/') || path.len() > 2048 || path.contains(['\0', '\r', '\n'])
117                {
118                    return Err("HTTP health path must be a bounded absolute request path".into());
119                }
120                if expected_statuses.is_empty()
121                    || expected_statuses.len() > 32
122                    || expected_statuses
123                        .iter()
124                        .any(|status| !(100..=599).contains(status))
125                {
126                    return Err("HTTP health expected_statuses are invalid".into());
127                }
128                Ok(())
129            }
130            Self::Tcp { port } => {
131                super::validate_name("health port", port)?;
132                if !network.has_port(port) {
133                    return Err(format!("TCP health check references unknown port {port:?}"));
134                }
135                Ok(())
136            }
137            Self::Command { command } => {
138                if command.is_empty() || command.len() > 64 {
139                    return Err("command health check requires 1 to 64 arguments".into());
140                }
141                for value in command {
142                    if value.is_empty() || value.len() > 32 * 1024 || value.contains('\0') {
143                        return Err("command health check contains an invalid argument".into());
144                    }
145                }
146                Ok(())
147            }
148        }
149    }
150}
151
152#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
153#[serde(deny_unknown_fields)]
154pub struct RuntimeHealthCheck {
155    pub probe: HealthProbe,
156    pub interval_ms: u64,
157    pub timeout_ms: u64,
158    pub start_period_ms: u64,
159    pub success_threshold: u32,
160    pub failure_threshold: u32,
161}
162
163impl RuntimeHealthCheck {
164    fn validate(&self, network: &RuntimeNetworkSpec) -> Result<(), String> {
165        if self.interval_ms == 0
166            || self.timeout_ms == 0
167            || self.timeout_ms > self.interval_ms
168            || self.success_threshold == 0
169            || self.failure_threshold == 0
170        {
171            return Err("health timing and threshold values are invalid".into());
172        }
173        self.probe.validate(network)
174    }
175}
176
177#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
178#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
179pub enum RestartPolicy {
180    Never,
181    OnFailure { max_retries: u32 },
182    Always,
183}
184
185#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
186#[serde(deny_unknown_fields)]
187pub struct RuntimeOutputSpec {
188    pub name: String,
189    pub path: String,
190    pub media_type: String,
191    pub max_bytes: u64,
192}
193
194impl RuntimeOutputSpec {
195    fn validate(&self) -> Result<(), String> {
196        super::validate_name("output name", &self.name)?;
197        super::validate_absolute_path("output path", &self.path)?;
198        super::validate_nonempty("output media_type", &self.media_type, 255)?;
199        if self.max_bytes == 0 {
200            return Err("output max_bytes must be positive".into());
201        }
202        Ok(())
203    }
204}
205
206/// Immutable provider-neutral definition of one finite Task or long-running
207/// Service generation.
208#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
209#[serde(deny_unknown_fields)]
210pub struct RuntimeUnitSpec {
211    pub schema: String,
212    pub unit_id: String,
213    pub generation: u64,
214    pub class: RuntimeUnitClass,
215    pub artifact: ArtifactRef,
216    pub process: RuntimeProcessSpec,
217    pub mounts: Vec<RuntimeMount>,
218    pub secrets: Vec<SecretReference>,
219    pub network: RuntimeNetworkSpec,
220    pub resources: ResourceLimits,
221    pub isolation: IsolationLevel,
222    pub health: Option<RuntimeHealthCheck>,
223    pub restart: RestartPolicy,
224    pub outputs: Vec<RuntimeOutputSpec>,
225    pub semantics_profile_digest: Option<String>,
226}
227
228impl RuntimeUnitSpec {
229    pub const SCHEMA: &'static str = "a3s.runtime.unit-spec.v2";
230
231    pub fn validate(&self) -> Result<(), String> {
232        if self.schema != Self::SCHEMA {
233            return Err(format!("unsupported Runtime unit schema {:?}", self.schema));
234        }
235        super::validate_id("unit_id", &self.unit_id, 512)?;
236        if self.generation == 0 {
237            return Err("Runtime unit generation must be positive".into());
238        }
239        self.artifact.validate()?;
240        self.process.validate()?;
241        self.network.validate()?;
242        self.resources.validate()?;
243        if self.mounts.len() > 128 || self.secrets.len() > 128 || self.outputs.len() > 128 {
244            return Err("Runtime unit input or output count exceeds protocol limits".into());
245        }
246
247        let mut mount_names = BTreeSet::new();
248        let mut mount_targets = BTreeSet::new();
249        for mount in &self.mounts {
250            mount.validate()?;
251            if !mount_names.insert(&mount.name) || !mount_targets.insert(&mount.target) {
252                return Err("Runtime mount names and targets must be unique".into());
253            }
254        }
255
256        let mut secret_names = BTreeSet::new();
257        let mut secret_targets = BTreeSet::new();
258        for secret in &self.secrets {
259            secret.validate()?;
260            let target = serde_json::to_string(&secret.target)
261                .map_err(|error| format!("could not encode secret target: {error}"))?;
262            if !secret_names.insert(&secret.name) || !secret_targets.insert(target) {
263                return Err("Runtime secret names and targets must be unique".into());
264            }
265        }
266
267        let mut output_names = BTreeSet::new();
268        let mut output_paths = BTreeSet::new();
269        for output in &self.outputs {
270            output.validate()?;
271            if !output_names.insert(&output.name) || !output_paths.insert(&output.path) {
272                return Err("Runtime output names and paths must be unique".into());
273            }
274        }
275
276        if let Some(digest) = &self.semantics_profile_digest {
277            super::validate_digest(digest)?;
278        }
279
280        match self.class {
281            RuntimeUnitClass::Task => {
282                if self.resources.execution_timeout_ms.is_none() {
283                    return Err("Task requires execution_timeout_ms".into());
284                }
285                if self.health.is_some() || matches!(self.restart, RestartPolicy::Always) {
286                    return Err("Task cannot use health checks or an always restart policy".into());
287                }
288            }
289            RuntimeUnitClass::Service => {
290                if self.resources.execution_timeout_ms.is_some() || !self.outputs.is_empty() {
291                    return Err("Service cannot use an execution timeout or Task outputs".into());
292                }
293                if let Some(health) = &self.health {
294                    health.validate(&self.network)?;
295                }
296            }
297        }
298        Ok(())
299    }
300
301    pub fn digest(&self) -> Result<String, String> {
302        self.validate()?;
303        let bytes = serde_json::to_vec(self)
304            .map_err(|error| format!("could not encode Runtime unit spec: {error}"))?;
305        Ok(format!("sha256:{:x}", Sha256::digest(bytes)))
306    }
307}
308
309#[cfg(test)]
310mod tests {
311    use super::*;
312    use std::collections::BTreeMap;
313
314    fn artifact() -> ArtifactRef {
315        ArtifactRef {
316            uri: format!("oci://registry.example/a3s/demo@sha256:{}", "a".repeat(64)),
317            digest: format!("sha256:{}", "a".repeat(64)),
318            media_type: "application/vnd.oci.image.manifest.v1+json".into(),
319        }
320    }
321
322    fn resources(timeout: Option<u64>) -> ResourceLimits {
323        ResourceLimits {
324            cpu_millis: 500,
325            memory_bytes: 128 * 1024 * 1024,
326            pids: 128,
327            ephemeral_storage_bytes: Some(1024 * 1024 * 1024),
328            execution_timeout_ms: timeout,
329        }
330    }
331
332    fn task() -> RuntimeUnitSpec {
333        RuntimeUnitSpec {
334            schema: RuntimeUnitSpec::SCHEMA.into(),
335            unit_id: "build-1".into(),
336            generation: 1,
337            class: RuntimeUnitClass::Task,
338            artifact: artifact(),
339            process: RuntimeProcessSpec {
340                command: vec!["/bin/build".into()],
341                args: vec![],
342                working_directory: Some("/workspace".into()),
343                environment: BTreeMap::new(),
344            },
345            mounts: vec![],
346            secrets: vec![],
347            network: RuntimeNetworkSpec {
348                mode: super::super::NetworkMode::Outbound,
349                ports: vec![],
350            },
351            resources: resources(Some(60_000)),
352            isolation: IsolationLevel::Container,
353            health: None,
354            restart: RestartPolicy::OnFailure { max_retries: 1 },
355            outputs: vec![RuntimeOutputSpec {
356                name: "image".into(),
357                path: "/outputs/image.json".into(),
358                media_type: "application/json".into(),
359                max_bytes: 1024,
360            }],
361            semantics_profile_digest: None,
362        }
363    }
364
365    fn service() -> RuntimeUnitSpec {
366        let mut spec = task();
367        spec.unit_id = "service-1".into();
368        spec.class = RuntimeUnitClass::Service;
369        spec.resources = resources(None);
370        spec.outputs.clear();
371        spec.restart = RestartPolicy::Always;
372        spec.network = RuntimeNetworkSpec {
373            mode: super::super::NetworkMode::Service,
374            ports: vec![super::super::RuntimePort {
375                name: "http".into(),
376                container_port: 8080,
377                protocol: super::super::TransportProtocol::Tcp,
378            }],
379        };
380        spec.health = Some(RuntimeHealthCheck {
381            probe: HealthProbe::Http {
382                port: "http".into(),
383                path: "/health".into(),
384                expected_statuses: vec![200],
385            },
386            interval_ms: 5_000,
387            timeout_ms: 1_000,
388            start_period_ms: 10_000,
389            success_threshold: 1,
390            failure_threshold: 3,
391        });
392        spec
393    }
394
395    #[test]
396    fn task_and_service_specs_are_general_and_digest_stable() {
397        let task = task();
398        let service = service();
399        task.validate().unwrap();
400        service.validate().unwrap();
401        assert_eq!(task.digest().unwrap(), task.digest().unwrap());
402        assert_ne!(task.digest().unwrap(), service.digest().unwrap());
403    }
404
405    #[test]
406    fn lifecycle_specific_fields_fail_closed() {
407        let mut task = task();
408        task.resources.execution_timeout_ms = None;
409        assert!(task.validate().is_err());
410
411        let mut service = service();
412        service.outputs.push(RuntimeOutputSpec {
413            name: "invalid".into(),
414            path: "/output".into(),
415            media_type: "text/plain".into(),
416            max_bytes: 1,
417        });
418        assert!(service.validate().is_err());
419    }
420
421    #[test]
422    fn health_checks_reference_declared_ports() {
423        let mut service = service();
424        let health = service.health.as_mut().unwrap();
425        health.probe = HealthProbe::Tcp {
426            port: "missing".into(),
427        };
428        assert!(service.validate().is_err());
429    }
430}
a3s_runtime/contract/unit.rs

a3s_runtime/contract/
unit.rs
