Skip to main content

a3s_runtime/contract/
protocol.rs

1use super::{RuntimeObservation, RuntimeUnitSpec};
2use serde::{Deserialize, Serialize};
3use sha2::{Digest, Sha256};
4
5#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
6#[serde(deny_unknown_fields)]
7pub struct RuntimeApplyRequest {
8    pub schema: String,
9    pub request_id: String,
10    pub deadline_at_ms: Option<u64>,
11    pub spec: RuntimeUnitSpec,
12}
13
14impl RuntimeApplyRequest {
15    pub const SCHEMA: &'static str = "a3s.runtime.apply-request.v1";
16
17    pub fn validate(&self) -> Result<(), String> {
18        if self.schema != Self::SCHEMA {
19            return Err(format!(
20                "unsupported Runtime apply schema {:?}",
21                self.schema
22            ));
23        }
24        super::validate_id("request_id", &self.request_id, 512)?;
25        if self.deadline_at_ms == Some(0) {
26            return Err("deadline_at_ms must be positive when present".into());
27        }
28        self.spec.validate()
29    }
30
31    pub fn digest(&self) -> Result<String, String> {
32        canonical_digest(self, self.validate())
33    }
34}
35
36#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
37#[serde(deny_unknown_fields)]
38pub struct RuntimeActionRequest {
39    pub schema: String,
40    pub request_id: String,
41    pub unit_id: String,
42    pub generation: u64,
43    pub deadline_at_ms: Option<u64>,
44}
45
46impl RuntimeActionRequest {
47    pub const SCHEMA: &'static str = "a3s.runtime.action-request.v1";
48
49    pub fn validate(&self) -> Result<(), String> {
50        if self.schema != Self::SCHEMA {
51            return Err(format!(
52                "unsupported Runtime action schema {:?}",
53                self.schema
54            ));
55        }
56        super::validate_id("request_id", &self.request_id, 512)?;
57        super::validate_id("unit_id", &self.unit_id, 512)?;
58        if self.generation == 0 || self.deadline_at_ms == Some(0) {
59            return Err("action generation and deadline must be positive".into());
60        }
61        Ok(())
62    }
63
64    pub fn digest(&self) -> Result<String, String> {
65        canonical_digest(self, self.validate())
66    }
67}
68
69#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
70#[serde(deny_unknown_fields)]
71pub struct RuntimeRemoval {
72    pub schema: String,
73    pub request_id: String,
74    pub unit_id: String,
75    pub generation: u64,
76    pub removed_at_ms: u64,
77    pub already_absent: bool,
78}
79
80impl RuntimeRemoval {
81    pub const SCHEMA: &'static str = "a3s.runtime.removal.v1";
82
83    pub fn validate(&self) -> Result<(), String> {
84        if self.schema != Self::SCHEMA {
85            return Err(format!(
86                "unsupported Runtime removal schema {:?}",
87                self.schema
88            ));
89        }
90        super::validate_id("request_id", &self.request_id, 512)?;
91        super::validate_id("unit_id", &self.unit_id, 512)?;
92        if self.generation == 0 {
93            return Err("removal generation must be positive".into());
94        }
95        Ok(())
96    }
97}
98
99#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
100#[serde(rename_all = "snake_case")]
101pub enum RuntimeLogStream {
102    Stdout,
103    Stderr,
104}
105
106#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
107#[serde(rename_all = "snake_case")]
108pub enum RuntimeLogDiscontinuityReason {
109    CursorLost,
110    SourceDisconnected,
111}
112
113#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
114#[serde(deny_unknown_fields)]
115pub struct RuntimeLogQuery {
116    pub schema: String,
117    pub unit_id: String,
118    pub generation: u64,
119    pub cursor: Option<String>,
120    pub limit: u32,
121    pub stream: Option<RuntimeLogStream>,
122}
123
124impl RuntimeLogQuery {
125    pub const SCHEMA: &'static str = "a3s.runtime.log-query.v1";
126
127    pub fn validate(&self) -> Result<(), String> {
128        if self.schema != Self::SCHEMA {
129            return Err(format!(
130                "unsupported Runtime log query schema {:?}",
131                self.schema
132            ));
133        }
134        super::validate_id("unit_id", &self.unit_id, 512)?;
135        if self.generation == 0 || self.limit == 0 || self.limit > 10_000 {
136            return Err("log generation or limit is invalid".into());
137        }
138        if self
139            .cursor
140            .as_ref()
141            .is_some_and(|value| value.is_empty() || value.len() > 1024 || value.contains('\0'))
142        {
143            return Err("log cursor is invalid".into());
144        }
145        Ok(())
146    }
147}
148
149#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
150#[serde(deny_unknown_fields)]
151pub struct RuntimeLogChunk {
152    pub schema: String,
153    pub cursor: String,
154    pub sequence: u64,
155    pub observed_at_ms: u64,
156    pub stream: RuntimeLogStream,
157    pub data: String,
158}
159
160impl RuntimeLogChunk {
161    pub const SCHEMA: &'static str = "a3s.runtime.log-chunk.v1";
162
163    pub fn validate(&self) -> Result<(), String> {
164        if self.schema != Self::SCHEMA {
165            return Err(format!(
166                "unsupported Runtime log chunk schema {:?}",
167                self.schema
168            ));
169        }
170        super::validate_nonempty("log cursor", &self.cursor, 1024)?;
171        if self.data.len() > 1024 * 1024 {
172            return Err("log chunk exceeds one MiB".into());
173        }
174        Ok(())
175    }
176}
177
178#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
179#[serde(deny_unknown_fields)]
180pub struct RuntimeExecRequest {
181    pub schema: String,
182    pub request_id: String,
183    pub unit_id: String,
184    pub generation: u64,
185    pub command: Vec<String>,
186    pub timeout_ms: u64,
187    pub deadline_at_ms: Option<u64>,
188}
189
190impl RuntimeExecRequest {
191    pub const SCHEMA: &'static str = "a3s.runtime.exec-request.v1";
192
193    pub fn validate(&self) -> Result<(), String> {
194        if self.schema != Self::SCHEMA {
195            return Err(format!(
196                "unsupported Runtime exec request schema {:?}",
197                self.schema
198            ));
199        }
200        super::validate_id("request_id", &self.request_id, 512)?;
201        super::validate_id("unit_id", &self.unit_id, 512)?;
202        if self.generation == 0
203            || self.timeout_ms == 0
204            || self.deadline_at_ms == Some(0)
205            || self.command.is_empty()
206            || self.command.len() > 256
207            || self
208                .command
209                .iter()
210                .any(|value| value.is_empty() || value.len() > 32 * 1024 || value.contains('\0'))
211        {
212            return Err("exec request is invalid".into());
213        }
214        Ok(())
215    }
216
217    pub fn digest(&self) -> Result<String, String> {
218        canonical_digest(self, self.validate())
219    }
220}
221
222#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
223#[serde(deny_unknown_fields)]
224pub struct RuntimeExecResult {
225    pub schema: String,
226    pub request_id: String,
227    pub observation: RuntimeObservation,
228    pub exit_code: i32,
229    pub stdout: String,
230    pub stderr: String,
231    pub truncated: bool,
232}
233
234impl RuntimeExecResult {
235    pub const SCHEMA: &'static str = "a3s.runtime.exec-result.v1";
236
237    pub fn validate(&self) -> Result<(), String> {
238        if self.schema != Self::SCHEMA {
239            return Err(format!(
240                "unsupported Runtime exec result schema {:?}",
241                self.schema
242            ));
243        }
244        super::validate_id("request_id", &self.request_id, 512)?;
245        self.observation.validate()?;
246        if self.stdout.len() > 16 * 1024 * 1024 || self.stderr.len() > 16 * 1024 * 1024 {
247            return Err("exec output exceeds protocol limits".into());
248        }
249        Ok(())
250    }
251}
252
253fn canonical_digest<T: Serialize>(value: &T, valid: Result<(), String>) -> Result<String, String> {
254    valid?;
255    let bytes = serde_json::to_vec(value)
256        .map_err(|error| format!("could not encode Runtime request: {error}"))?;
257    Ok(format!("sha256:{:x}", Sha256::digest(bytes)))
258}
259
260#[cfg(test)]
261mod tests {
262    use super::*;
263
264    #[test]
265    fn log_discontinuity_reasons_have_stable_wire_names() {
266        assert_eq!(
267            serde_json::to_string(&RuntimeLogDiscontinuityReason::CursorLost)
268                .expect("serialize cursor loss"),
269            "\"cursor_lost\""
270        );
271        assert_eq!(
272            serde_json::to_string(&RuntimeLogDiscontinuityReason::SourceDisconnected)
273                .expect("serialize source disconnect"),
274            "\"source_disconnected\""
275        );
276    }
277}