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/// Optional Service lifecycle policy that is distinct from readiness.
178///
179/// Providers advertising `ServiceLifecycle` sample the liveness probe and
180/// honor the declared graceful-stop deadline before forcing termination.
181#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
182#[serde(deny_unknown_fields)]
183pub struct RuntimeServiceLifecycle {
184    pub liveness: RuntimeHealthCheck,
185    pub shutdown_grace_seconds: u32,
186}
187
188impl RuntimeServiceLifecycle {
189    fn validate(&self, network: &RuntimeNetworkSpec) -> Result<(), String> {
190        self.liveness.validate(network)?;
191        if !(1..=3_600).contains(&self.shutdown_grace_seconds) {
192            return Err("Service shutdown_grace_seconds must be between 1 and 3600".into());
193        }
194        Ok(())
195    }
196}
197
198#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
199#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
200pub enum RestartPolicy {
201    Never,
202    OnFailure { max_retries: u32 },
203    Always,
204}
205
206#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
207#[serde(deny_unknown_fields)]
208pub struct RuntimeOutputSpec {
209    pub name: String,
210    pub path: String,
211    pub media_type: String,
212    pub max_bytes: u64,
213}
214
215impl RuntimeOutputSpec {
216    fn validate(&self) -> Result<(), String> {
217        super::validate_name("output name", &self.name)?;
218        super::validate_absolute_path("output path", &self.path)?;
219        super::validate_nonempty("output media_type", &self.media_type, 255)?;
220        if self.max_bytes == 0 {
221            return Err("output max_bytes must be positive".into());
222        }
223        Ok(())
224    }
225}
226
227/// Immutable provider-neutral definition of one finite Task or long-running
228/// Service generation.
229#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
230#[serde(deny_unknown_fields)]
231pub struct RuntimeUnitSpec {
232    pub schema: String,
233    pub unit_id: String,
234    pub generation: u64,
235    pub class: RuntimeUnitClass,
236    pub artifact: ArtifactRef,
237    pub process: RuntimeProcessSpec,
238    pub mounts: Vec<RuntimeMount>,
239    pub secrets: Vec<SecretReference>,
240    pub network: RuntimeNetworkSpec,
241    pub resources: ResourceLimits,
242    pub isolation: IsolationLevel,
243    /// Readiness policy used to decide whether a running Service can receive
244    /// traffic. Liveness and graceful stop are declared separately.
245    pub health: Option<RuntimeHealthCheck>,
246    pub service_lifecycle: Option<RuntimeServiceLifecycle>,
247    pub restart: RestartPolicy,
248    pub outputs: Vec<RuntimeOutputSpec>,
249    pub semantics_profile_digest: Option<String>,
250    pub identity_attachment_digest: Option<String>,
251}
252
253impl RuntimeUnitSpec {
254    pub const SCHEMA: &'static str = "a3s.runtime.unit-spec.v4";
255
256    pub fn validate(&self) -> Result<(), String> {
257        if self.schema != Self::SCHEMA {
258            return Err(format!("unsupported Runtime unit schema {:?}", self.schema));
259        }
260        super::validate_id("unit_id", &self.unit_id, 512)?;
261        if self.generation == 0 {
262            return Err("Runtime unit generation must be positive".into());
263        }
264        self.artifact.validate()?;
265        self.process.validate()?;
266        self.network.validate()?;
267        self.resources.validate()?;
268        if self.mounts.len() > 128 || self.secrets.len() > 128 || self.outputs.len() > 128 {
269            return Err("Runtime unit input or output count exceeds protocol limits".into());
270        }
271
272        let mut mount_names = BTreeSet::new();
273        let mut mount_targets = BTreeSet::new();
274        for mount in &self.mounts {
275            mount.validate()?;
276            if !mount_names.insert(&mount.name) || !mount_targets.insert(&mount.target) {
277                return Err("Runtime mount names and targets must be unique".into());
278            }
279        }
280
281        let mut secret_names = BTreeSet::new();
282        let mut secret_targets = BTreeSet::new();
283        for secret in &self.secrets {
284            secret.validate()?;
285            let target = serde_json::to_string(&secret.target)
286                .map_err(|error| format!("could not encode secret target: {error}"))?;
287            if !secret_names.insert(&secret.name) || !secret_targets.insert(target) {
288                return Err("Runtime secret names and targets must be unique".into());
289            }
290        }
291
292        let mut output_names = BTreeSet::new();
293        let mut output_paths = BTreeSet::new();
294        for output in &self.outputs {
295            output.validate()?;
296            if !output_names.insert(&output.name) || !output_paths.insert(&output.path) {
297                return Err("Runtime output names and paths must be unique".into());
298            }
299        }
300
301        if let Some(digest) = &self.semantics_profile_digest {
302            super::validate_digest(digest)?;
303        }
304        if let Some(digest) = &self.identity_attachment_digest {
305            super::validate_digest(digest)?;
306        }
307
308        match self.class {
309            RuntimeUnitClass::Task => {
310                if self.resources.execution_timeout_ms.is_none() {
311                    return Err("Task requires execution_timeout_ms".into());
312                }
313                if self.health.is_some()
314                    || self.service_lifecycle.is_some()
315                    || matches!(self.restart, RestartPolicy::Always)
316                {
317                    return Err(
318                        "Task cannot use Service health, lifecycle, or an always restart policy"
319                            .into(),
320                    );
321                }
322            }
323            RuntimeUnitClass::Service => {
324                if self.resources.execution_timeout_ms.is_some() || !self.outputs.is_empty() {
325                    return Err("Service cannot use an execution timeout or Task outputs".into());
326                }
327                if let Some(health) = &self.health {
328                    health.validate(&self.network)?;
329                }
330                if let Some(lifecycle) = &self.service_lifecycle {
331                    if self.health.is_none() {
332                        return Err(
333                            "Service lifecycle requires a separate readiness health policy".into(),
334                        );
335                    }
336                    lifecycle.validate(&self.network)?;
337                }
338            }
339        }
340        Ok(())
341    }
342
343    pub fn digest(&self) -> Result<String, String> {
344        self.validate()?;
345        let bytes = serde_json::to_vec(self)
346            .map_err(|error| format!("could not encode Runtime unit spec: {error}"))?;
347        Ok(format!("sha256:{:x}", Sha256::digest(bytes)))
348    }
349}
350
351#[cfg(test)]
352mod tests {
353    use super::*;
354    use std::collections::BTreeMap;
355
356    fn artifact() -> ArtifactRef {
357        ArtifactRef {
358            uri: format!("oci://registry.example/a3s/demo@sha256:{}", "a".repeat(64)),
359            digest: format!("sha256:{}", "a".repeat(64)),
360            media_type: "application/vnd.oci.image.manifest.v1+json".into(),
361        }
362    }
363
364    fn resources(timeout: Option<u64>) -> ResourceLimits {
365        ResourceLimits {
366            cpu_millis: 500,
367            memory_bytes: 128 * 1024 * 1024,
368            pids: 128,
369            ephemeral_storage_bytes: Some(1024 * 1024 * 1024),
370            execution_timeout_ms: timeout,
371        }
372    }
373
374    fn task() -> RuntimeUnitSpec {
375        RuntimeUnitSpec {
376            schema: RuntimeUnitSpec::SCHEMA.into(),
377            unit_id: "build-1".into(),
378            generation: 1,
379            class: RuntimeUnitClass::Task,
380            artifact: artifact(),
381            process: RuntimeProcessSpec {
382                command: vec!["/bin/build".into()],
383                args: vec![],
384                working_directory: Some("/workspace".into()),
385                environment: BTreeMap::new(),
386            },
387            mounts: vec![],
388            secrets: vec![],
389            network: RuntimeNetworkSpec {
390                mode: super::super::NetworkMode::Outbound,
391                ports: vec![],
392            },
393            resources: resources(Some(60_000)),
394            isolation: IsolationLevel::Container,
395            health: None,
396            service_lifecycle: None,
397            restart: RestartPolicy::OnFailure { max_retries: 1 },
398            outputs: vec![RuntimeOutputSpec {
399                name: "image".into(),
400                path: "/outputs/image.json".into(),
401                media_type: "application/json".into(),
402                max_bytes: 1024,
403            }],
404            semantics_profile_digest: None,
405            identity_attachment_digest: None,
406        }
407    }
408
409    fn service() -> RuntimeUnitSpec {
410        let mut spec = task();
411        spec.unit_id = "service-1".into();
412        spec.class = RuntimeUnitClass::Service;
413        spec.resources = resources(None);
414        spec.outputs.clear();
415        spec.restart = RestartPolicy::Always;
416        spec.network = RuntimeNetworkSpec {
417            mode: super::super::NetworkMode::Service,
418            ports: vec![super::super::RuntimePort {
419                name: "http".into(),
420                container_port: 8080,
421                protocol: super::super::TransportProtocol::Tcp,
422            }],
423        };
424        spec.health = Some(RuntimeHealthCheck {
425            probe: HealthProbe::Http {
426                port: "http".into(),
427                path: "/health".into(),
428                expected_statuses: vec![200],
429            },
430            interval_ms: 5_000,
431            timeout_ms: 1_000,
432            start_period_ms: 10_000,
433            success_threshold: 1,
434            failure_threshold: 3,
435        });
436        spec
437    }
438
439    #[test]
440    fn task_and_service_specs_are_general_and_digest_stable() {
441        let task = task();
442        let service = service();
443        task.validate().unwrap();
444        service.validate().unwrap();
445        assert_eq!(task.digest().unwrap(), task.digest().unwrap());
446        assert_ne!(task.digest().unwrap(), service.digest().unwrap());
447    }
448
449    #[test]
450    fn lifecycle_specific_fields_fail_closed() {
451        let mut task = task();
452        task.resources.execution_timeout_ms = None;
453        assert!(task.validate().is_err());
454
455        let mut service = service();
456        service.outputs.push(RuntimeOutputSpec {
457            name: "invalid".into(),
458            path: "/output".into(),
459            media_type: "text/plain".into(),
460            max_bytes: 1,
461        });
462        assert!(service.validate().is_err());
463    }
464
465    #[test]
466    fn health_checks_reference_declared_ports() {
467        let mut service = service();
468        let health = service.health.as_mut().unwrap();
469        health.probe = HealthProbe::Tcp {
470            port: "missing".into(),
471        };
472        assert!(service.validate().is_err());
473    }
474
475    #[test]
476    fn service_lifecycle_is_service_only_readiness_bound_and_bounded() {
477        let mut service = service();
478        service.service_lifecycle = Some(RuntimeServiceLifecycle {
479            liveness: RuntimeHealthCheck {
480                probe: HealthProbe::Http {
481                    port: "http".into(),
482                    path: "/health/live".into(),
483                    expected_statuses: vec![200],
484                },
485                interval_ms: 5_000,
486                timeout_ms: 1_000,
487                start_period_ms: 10_000,
488                success_threshold: 1,
489                failure_threshold: 3,
490            },
491            shutdown_grace_seconds: 30,
492        });
493        service.validate().expect("Service lifecycle");
494
495        service.health = None;
496        assert!(service.validate().is_err());
497        service.health = Some(RuntimeHealthCheck {
498            probe: HealthProbe::Tcp {
499                port: "http".into(),
500            },
501            interval_ms: 5_000,
502            timeout_ms: 1_000,
503            start_period_ms: 0,
504            success_threshold: 1,
505            failure_threshold: 3,
506        });
507        service
508            .service_lifecycle
509            .as_mut()
510            .expect("lifecycle")
511            .shutdown_grace_seconds = 0;
512        assert!(service.validate().is_err());
513        service
514            .service_lifecycle
515            .as_mut()
516            .expect("lifecycle")
517            .shutdown_grace_seconds = 1;
518        service.validate().expect("minimum shutdown grace");
519        service
520            .service_lifecycle
521            .as_mut()
522            .expect("lifecycle")
523            .shutdown_grace_seconds = 3_600;
524        service.validate().expect("maximum shutdown grace");
525        service
526            .service_lifecycle
527            .as_mut()
528            .expect("lifecycle")
529            .shutdown_grace_seconds = 3_601;
530        assert!(service.validate().is_err());
531
532        let mut task = task();
533        task.service_lifecycle = service.service_lifecycle;
534        assert!(task.validate().is_err());
535    }
536
537    #[test]
538    fn registry_credential_is_a_typed_unique_secret_target() {
539        let mut service = service();
540        service.secrets.push(SecretReference {
541            name: "registry".into(),
542            reference: "secret://registry/7".into(),
543            target: super::super::SecretTarget::RegistryCredential,
544        });
545        service.validate().expect("registry credential");
546        assert_eq!(
547            serde_json::to_value(&service.secrets[0].target).expect("target JSON"),
548            serde_json::json!({"kind": "registry_credential"})
549        );
550
551        service.secrets.push(SecretReference {
552            name: "other-registry".into(),
553            reference: "secret://registry/8".into(),
554            target: super::super::SecretTarget::RegistryCredential,
555        });
556        assert!(service.validate().is_err());
557    }
558}
a3s_runtime/contract/unit.rs

a3s_runtime/contract/
unit.rs
