microsandbox_protocol/
jobs.rs

1//! Additive host-control contract for runtime-owned command jobs.
2//!
3//! This extension deliberately does not change guest exec messages or the released control enums.
4//! Clients first request `Hello`, then fence every operation with the returned runtime identity.
5
6use serde::{Deserialize, Serialize};
7
8use crate::exec::{ExecFailed, ExecRequest};
9
10//--------------------------------------------------------------------------------------------------
11// Constants
12//--------------------------------------------------------------------------------------------------
13
14/// First version of the managed-job contract and persisted metadata.
15pub const JOB_PROTOCOL_VERSION: u32 = 1;
16/// Capability-probed host-control extension name.
17pub const JOB_REQUEST: &str = "control.jobs";
18/// Terminal response to one bounded job operation.
19pub const JOB_RESPONSE: &str = "control.jobs.result";
20/// Maximum bytes accepted in one input write or returned in one output chunk.
21pub const JOB_CHUNK_BYTES: usize = 16 * 1024;
22/// Maximum simultaneously active jobs in one sandbox.
23pub const MAX_ACTIVE_JOBS: usize = 32;
24/// Maximum retained job records, including active jobs.
25pub const MAX_RETAINED_JOBS: usize = 256;
26/// Maximum encoded metadata in a listing page, independently of its item limit.
27pub const JOB_LIST_BYTES: usize = 256 * 1024;
28/// Reserved reply capacity for one job operation, including metadata and output chunks.
29pub const JOB_REPLY_BYTES: u32 = 512 * 1024;
30
31//--------------------------------------------------------------------------------------------------
32// Types
33//--------------------------------------------------------------------------------------------------
34
35/// An opaque job identity, distinct from guest PIDs and connection-local exec IDs.
36#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
37#[serde(try_from = "String", into = "String")]
38pub struct JobId(String);
39
40/// Persisted lifecycle observation. A lost runtime never implies a successful exit.
41#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
42#[serde(rename_all = "snake_case")]
43#[non_exhaustive]
44pub enum JobState {
45    /// Admission is durable; guest spawn has not yet been confirmed.
46    Starting,
47    /// Guest spawn was confirmed. Sandbox pause is orthogonal to this state.
48    Running,
49    /// The guest reported process completion.
50    Exited,
51    /// The guest rejected the command before it started.
52    Failed,
53    /// Ownership or transport was lost without a confirmed exit result.
54    Lost,
55}
56
57/// Durable job metadata. Environment values and stdin bytes are never included.
58#[derive(Debug, Clone, Serialize, Deserialize)]
59#[non_exhaustive]
60pub struct JobInfo {
61    /// Metadata format version.
62    pub version: u32,
63    /// Stable identity within the owning sandbox.
64    pub id: JobId,
65    /// Runtime generation that created this job.
66    pub runtime_boot_id: String,
67    /// Program followed by its arguments.
68    pub command: Vec<String>,
69    /// Most recent lifecycle observation.
70    pub state: JobState,
71    /// Whether output is a merged PTY stream.
72    pub tty: bool,
73    /// Whether further input has been permanently disabled.
74    pub stdin_closed: bool,
75    /// Guest PID, if spawn was acknowledged.
76    pub pid: Option<u32>,
77    /// Admission time in Unix milliseconds.
78    pub created_at: i64,
79    /// Confirmed spawn time in Unix milliseconds.
80    pub started_at: Option<i64>,
81    /// Completion or loss observation time in Unix milliseconds.
82    pub finished_at: Option<i64>,
83    /// Confirmed guest exit code, never synthesized on transport loss.
84    pub exit_code: Option<i32>,
85    /// Structured spawn failure, when supplied by the guest.
86    pub failure: Option<ExecFailed>,
87    /// Diagnostic for unconfirmed completion or host I/O failure.
88    pub error: Option<String>,
89    /// Whether the runtime requested termination after the execution deadline.
90    pub timed_out: bool,
91}
92
93/// Byte-preserving output record. Sequence numbers remain monotonic across log rotation.
94#[derive(Debug, Clone, Serialize, Deserialize)]
95#[non_exhaustive]
96pub struct JobOutput {
97    /// Monotonic sequence number, starting at one.
98    pub sequence: u64,
99    /// Unix milliseconds at capture.
100    pub timestamp: i64,
101    /// `stdout`, `stderr`, `output` (PTY), or `stdin_error`.
102    pub source: String,
103    /// Original bytes, encoded explicitly for portable persisted JSON.
104    pub data_base64: String,
105}
106
107/// One bounded list page.
108#[derive(Debug, Clone, Serialize, Deserialize)]
109pub struct JobPage {
110    /// Job summaries in stable identity order.
111    pub items: Vec<JobInfo>,
112    /// Last identity in this page when more records remain.
113    pub next_cursor: Option<JobId>,
114}
115
116/// One bounded replay result; missing retained output is reported explicitly.
117#[derive(Debug, Clone, Serialize, Deserialize)]
118pub struct JobOutputPage {
119    /// Captured records after the requested cursor.
120    pub items: Vec<JobOutput>,
121    /// Cursor after the returned records, or the requested cursor when empty.
122    pub cursor: u64,
123    /// Earliest available sequence when requested records were pruned.
124    pub gap_before: Option<u64>,
125    /// Latest metadata, including terminal state after output has been captured.
126    pub info: JobInfo,
127}
128
129/// One capability-probed, runtime-fenced request.
130#[derive(Debug, Clone, Serialize, Deserialize)]
131pub struct JobRequest {
132    /// Expected extension version.
133    pub version: u32,
134    /// Required for every operation except Hello.
135    pub runtime_boot_id: Option<String>,
136    /// Requested bounded operation.
137    pub operation: JobOperation,
138}
139
140/// Host-only operations. None waits for a process to exit or for output to become available.
141#[derive(Debug, Clone, Serialize, Deserialize)]
142#[serde(tag = "op", rename_all = "snake_case")]
143#[non_exhaustive]
144pub enum JobOperation {
145    /// Discover support without creating a process.
146    Hello,
147    /// Admit a launch exactly once for this ID and command fingerprint.
148    Start {
149        /// Client-generated random identity and idempotency key.
150        id: JobId,
151        /// Existing guest command contract.
152        command: ExecRequest,
153        /// Close guest stdin immediately after confirmed spawn.
154        no_stdin: bool,
155        /// Optional finite pipe input (at most 64 KiB), followed by EOF.
156        input_base64: Option<String>,
157        /// Host-enforced elapsed execution limit in milliseconds.
158        timeout_ms: Option<u64>,
159    },
160    /// List active or all retained jobs.
161    List {
162        /// Include terminal records.
163        all: bool,
164        /// Maximum page size (1..=100).
165        limit: usize,
166        /// Continue after this identity.
167        cursor: Option<JobId>,
168    },
169    /// Inspect one retained job.
170    Inspect {
171        /// Owning job identity.
172        id: JobId,
173    },
174    /// Read bounded output, optionally renewing an attachment lease.
175    Read {
176        /// Owning job identity.
177        id: JobId,
178        /// Resume strictly after this output sequence.
179        after: u64,
180        /// Opaque attachment lease token.
181        attachment: Option<String>,
182    },
183    /// Acquire one attachment. Reusing the same token is idempotent.
184    Attach {
185        /// Owning job identity.
186        id: JobId,
187        /// Opaque attachment lease token.
188        attachment: String,
189        /// Observe without acquiring input ownership.
190        read_only: bool,
191    },
192    /// Release an attachment without EOF or process termination.
193    Detach {
194        /// Owning job identity.
195        id: JobId,
196        /// Lease to release.
197        attachment: String,
198    },
199    /// Keep an attachment alive without consuming output.
200    Renew {
201        /// Owning job identity.
202        id: JobId,
203        /// Lease to renew.
204        attachment: String,
205    },
206    /// Queue input for the current input-owning attachment.
207    Write {
208        /// Owning job identity.
209        id: JobId,
210        /// Opaque attachment lease token.
211        attachment: String,
212        /// Original input bytes encoded as base64.
213        data_base64: String,
214    },
215    /// Resize the terminal for the input-owning attachment.
216    Resize {
217        /// Owning job identity.
218        id: JobId,
219        /// Opaque attachment lease token.
220        attachment: String,
221        /// Nonzero terminal row count.
222        rows: u16,
223        /// Nonzero terminal column count.
224        cols: u16,
225    },
226    /// Send a Linux guest process-group signal.
227    Signal {
228        /// Owning job identity.
229        id: JobId,
230        /// Linux guest signal number.
231        signal: i32,
232    },
233    /// Close a pipe after already admitted input; invalid for PTYs.
234    Eof {
235        /// Owning job identity.
236        id: JobId,
237    },
238}
239
240/// Result of one operation. Errors retain machine-readable codes across SDKs.
241#[derive(Debug, Clone, Serialize, Deserialize)]
242#[serde(tag = "result", rename_all = "snake_case")]
243#[non_exhaustive]
244pub enum JobResponse {
245    /// Successful capability discovery.
246    Hello {
247        /// Selected managed-job protocol version.
248        version: u32,
249        /// Immutable identity of the selected runtime.
250        runtime_boot_id: String,
251    },
252    /// Admission or inspection result.
253    Info {
254        /// Current job metadata.
255        info: JobInfo,
256    },
257    /// List page.
258    Page {
259        /// Bounded list result.
260        page: JobPage,
261    },
262    /// Output page.
263    Output {
264        /// Bounded output result.
265        page: JobOutputPage,
266    },
267    /// Attachment metadata and current replay cursor.
268    Attached {
269        /// Current job metadata.
270        info: JobInfo,
271        /// Latest captured sequence at admission.
272        cursor: u64,
273    },
274    /// Control operation was admitted.
275    Ok,
276    /// Explicit refusal or failure; never a fabricated process exit.
277    Error {
278        /// Stable machine-readable code.
279        code: String,
280        /// Human-readable diagnostic.
281        message: String,
282    },
283}
284
285//--------------------------------------------------------------------------------------------------
286// Methods
287//--------------------------------------------------------------------------------------------------
288
289impl JobId {
290    /// Validate the portable opaque ID, including before it becomes a path component.
291    pub fn parse(value: impl Into<String>) -> Result<Self, String> {
292        let value = value.into();
293        if value.len() != 36
294            || !value.starts_with("job_")
295            || !value[4..]
296                .bytes()
297                .all(|b| b.is_ascii_hexdigit() && !b.is_ascii_uppercase())
298        {
299            return Err("job ID must be job_ followed by 32 lowercase hexadecimal digits".into());
300        }
301        Ok(Self(value))
302    }
303
304    /// Return the stable string representation.
305    pub fn as_str(&self) -> &str {
306        &self.0
307    }
308}
309
310impl JobState {
311    /// Whether the runtime may still own a live process for this record.
312    pub fn is_active(self) -> bool {
313        matches!(self, Self::Starting | Self::Running)
314    }
315}
316
317impl JobInfo {
318    /// Construct a pending admission without persisting the command's environment or input.
319    pub fn starting(
320        id: JobId,
321        runtime_boot_id: String,
322        command: &ExecRequest,
323        no_stdin: bool,
324        created_at: i64,
325    ) -> Self {
326        Self {
327            version: JOB_PROTOCOL_VERSION,
328            id,
329            runtime_boot_id,
330            command: std::iter::once(command.cmd.clone())
331                .chain(command.args.clone())
332                .collect(),
333            state: JobState::Starting,
334            tty: command.tty,
335            stdin_closed: no_stdin,
336            pid: None,
337            created_at,
338            started_at: None,
339            finished_at: None,
340            exit_code: None,
341            failure: None,
342            error: None,
343            timed_out: false,
344        }
345    }
346}
347
348impl JobOutput {
349    /// Construct one captured output record.
350    pub fn new(sequence: u64, timestamp: i64, source: String, data_base64: String) -> Self {
351        Self {
352            sequence,
353            timestamp,
354            source,
355            data_base64,
356        }
357    }
358}
359
360impl JobResponse {
361    /// Construct a structured failure without exposing request contents.
362    pub fn error(code: impl Into<String>, message: impl Into<String>) -> Self {
363        Self::Error {
364            code: code.into(),
365            message: message.into(),
366        }
367    }
368}
369
370impl JobPage {
371    /// Page sorted, filtered records by both count and encoded size.
372    pub fn bounded(
373        records: impl IntoIterator<Item = JobInfo>,
374        limit: usize,
375    ) -> Result<Self, String> {
376        let mut items: Vec<JobInfo> = Vec::new();
377        let mut bytes = 0;
378        let mut encoded = Vec::new();
379        for info in records {
380            encoded.clear();
381            ciborium::ser::into_writer(&info, &mut encoded).map_err(|error| error.to_string())?;
382            let size = encoded.len();
383            if limit == 0 || size > JOB_LIST_BYTES {
384                return Err("invalid page limit or oversized job metadata".into());
385            }
386            if !items.is_empty() && (items.len() >= limit || bytes + size > JOB_LIST_BYTES) {
387                return Ok(Self {
388                    next_cursor: items.last().map(|info| info.id.clone()),
389                    items,
390                });
391            }
392            bytes += size;
393            items.push(info);
394        }
395        Ok(Self {
396            items,
397            next_cursor: None,
398        })
399    }
400}
401
402//--------------------------------------------------------------------------------------------------
403// Trait Implementations
404//--------------------------------------------------------------------------------------------------
405
406impl TryFrom<String> for JobId {
407    type Error = String;
408    fn try_from(value: String) -> Result<Self, Self::Error> {
409        Self::parse(value)
410    }
411}
412impl From<JobId> for String {
413    fn from(value: JobId) -> Self {
414        value.0
415    }
416}
417impl AsRef<str> for JobId {
418    fn as_ref(&self) -> &str {
419        self.as_str()
420    }
421}
422impl std::fmt::Display for JobId {
423    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
424        f.write_str(&self.0)
425    }
426}
427
428//--------------------------------------------------------------------------------------------------
429// Tests
430//--------------------------------------------------------------------------------------------------
431
432#[cfg(test)]
433mod tests {
434    use super::*;
435
436    #[test]
437    fn job_ids_cannot_escape_their_storage_directory() {
438        for invalid in [
439            "../outside",
440            "job_../../../../../../../../../../..",
441            "job_ABCDEF",
442            "job_0000000000000000000000000000000/",
443        ] {
444            assert!(JobId::parse(invalid).is_err());
445            assert!(serde_json::from_value::<JobId>(serde_json::json!(invalid)).is_err());
446        }
447        assert!(JobId::parse("job_0123456789abcdef0123456789abcdef").is_ok());
448    }
449
450    #[test]
451    fn job_pages_are_bounded_by_encoded_bytes_as_well_as_count() {
452        let command: ExecRequest =
453            serde_json::from_value(serde_json::json!({"cmd": "x".repeat(16 * 1024)})).unwrap();
454        let records = (0..100).map(|n| {
455            JobInfo::starting(
456                JobId::parse(format!("job_{n:032x}")).unwrap(),
457                "boot".into(),
458                &command,
459                false,
460                0,
461            )
462        });
463        let page = JobPage::bounded(records, 100).unwrap();
464        assert!(page.items.len() < 100);
465        assert_eq!(
466            page.next_cursor.as_ref(),
467            page.items.last().map(|info| &info.id)
468        );
469        assert!(serde_json::to_vec(&page).unwrap().len() < JOB_REPLY_BYTES as usize);
470    }
471}
microsandbox_protocol/jobs.rs

microsandbox_protocol/
jobs.rs
