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/// Origin of one captured output record.
94#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
95#[serde(rename_all = "snake_case")]
96pub enum JobOutputSource {
97    /// Command stdout.
98    Stdout,
99    /// Command stderr.
100    Stderr,
101    /// Combined PTY stream.
102    Output,
103    /// Guest notice that a stdin write failed; not command output.
104    StdinError,
105}
106
107/// Byte-preserving output record. Sequence numbers remain monotonic across log rotation.
108#[derive(Debug, Clone, Serialize, Deserialize)]
109#[non_exhaustive]
110pub struct JobOutput {
111    /// Monotonic sequence number, starting at one.
112    pub sequence: u64,
113    /// Unix milliseconds at capture.
114    pub timestamp: i64,
115    /// Where the bytes came from.
116    pub source: JobOutputSource,
117    /// Original bytes, encoded explicitly for portable persisted JSON.
118    pub data_base64: String,
119}
120
121/// One bounded list page.
122#[derive(Debug, Clone, Serialize, Deserialize)]
123pub struct JobPage {
124    /// Job summaries in stable identity order.
125    pub items: Vec<JobInfo>,
126    /// Last identity in this page when more records remain.
127    pub next_cursor: Option<JobId>,
128}
129
130/// One bounded replay result; missing retained output is reported explicitly.
131#[derive(Debug, Clone, Serialize, Deserialize)]
132pub struct JobOutputPage {
133    /// Captured records after the requested cursor.
134    pub items: Vec<JobOutput>,
135    /// Cursor after the returned records, or the requested cursor when empty.
136    pub cursor: u64,
137    /// Earliest available sequence when requested records were pruned.
138    pub gap_before: Option<u64>,
139    /// Latest metadata, including terminal state after output has been captured.
140    pub info: JobInfo,
141}
142
143/// One capability-probed, runtime-fenced request.
144#[derive(Debug, Clone, Serialize, Deserialize)]
145pub struct JobRequest {
146    /// Expected extension version.
147    pub version: u32,
148    /// Required for every operation except Hello.
149    pub runtime_boot_id: Option<String>,
150    /// Requested bounded operation.
151    pub operation: JobOperation,
152}
153
154/// Host-only operations. None waits for a process to exit or for output to become available.
155#[derive(Debug, Clone, Serialize, Deserialize)]
156#[serde(tag = "op", rename_all = "snake_case")]
157#[non_exhaustive]
158pub enum JobOperation {
159    /// Discover support without creating a process.
160    Hello,
161    /// Admit a launch exactly once for this ID and command fingerprint.
162    Start {
163        /// Client-generated random identity and idempotency key.
164        id: JobId,
165        /// Existing guest command contract.
166        command: ExecRequest,
167        /// Close guest stdin immediately after confirmed spawn.
168        no_stdin: bool,
169        /// Optional finite pipe input (at most 64 KiB), followed by EOF.
170        input_base64: Option<String>,
171        /// Host-enforced elapsed execution limit in milliseconds.
172        timeout_ms: Option<u64>,
173    },
174    /// List active or all retained jobs.
175    List {
176        /// Include terminal records.
177        all: bool,
178        /// Maximum page size (1..=100).
179        limit: usize,
180        /// Continue after this identity.
181        cursor: Option<JobId>,
182    },
183    /// Inspect one retained job.
184    Inspect {
185        /// Owning job identity.
186        id: JobId,
187    },
188    /// Read bounded output, optionally renewing an attachment lease.
189    Read {
190        /// Owning job identity.
191        id: JobId,
192        /// Resume strictly after this output sequence.
193        after: u64,
194        /// Opaque attachment lease token.
195        attachment: Option<String>,
196    },
197    /// Acquire one attachment. Reusing the same token is idempotent.
198    Attach {
199        /// Owning job identity.
200        id: JobId,
201        /// Opaque attachment lease token.
202        attachment: String,
203        /// Observe without acquiring input ownership.
204        read_only: bool,
205    },
206    /// Release an attachment without EOF or process termination.
207    Detach {
208        /// Owning job identity.
209        id: JobId,
210        /// Lease to release.
211        attachment: String,
212    },
213    /// Keep an attachment alive without consuming output.
214    Renew {
215        /// Owning job identity.
216        id: JobId,
217        /// Lease to renew.
218        attachment: String,
219    },
220    /// Queue input for the current input-owning attachment.
221    Write {
222        /// Owning job identity.
223        id: JobId,
224        /// Opaque attachment lease token.
225        attachment: String,
226        /// Original input bytes encoded as base64.
227        data_base64: String,
228    },
229    /// Resize the terminal for the input-owning attachment.
230    Resize {
231        /// Owning job identity.
232        id: JobId,
233        /// Opaque attachment lease token.
234        attachment: String,
235        /// Nonzero terminal row count.
236        rows: u16,
237        /// Nonzero terminal column count.
238        cols: u16,
239    },
240    /// Send a Linux guest process-group signal.
241    Signal {
242        /// Owning job identity.
243        id: JobId,
244        /// Linux guest signal number.
245        signal: i32,
246    },
247    /// Close a pipe after already admitted input; invalid for PTYs.
248    Eof {
249        /// Owning job identity.
250        id: JobId,
251    },
252}
253
254/// Result of one operation. Errors retain machine-readable codes across SDKs.
255#[derive(Debug, Clone, Serialize, Deserialize)]
256#[serde(tag = "result", rename_all = "snake_case")]
257#[non_exhaustive]
258pub enum JobResponse {
259    /// Successful capability discovery.
260    Hello {
261        /// Selected managed-job protocol version.
262        version: u32,
263        /// Immutable identity of the selected runtime.
264        runtime_boot_id: String,
265    },
266    /// Admission or inspection result.
267    Info {
268        /// Current job metadata.
269        info: JobInfo,
270    },
271    /// List page.
272    Page {
273        /// Bounded list result.
274        page: JobPage,
275    },
276    /// Output page.
277    Output {
278        /// Bounded output result.
279        page: JobOutputPage,
280    },
281    /// Attachment metadata and current replay cursor.
282    Attached {
283        /// Current job metadata.
284        info: JobInfo,
285        /// Latest captured sequence at admission.
286        cursor: u64,
287    },
288    /// Control operation was admitted.
289    Ok,
290    /// Explicit refusal or failure; never a fabricated process exit.
291    Error {
292        /// Stable machine-readable code.
293        code: String,
294        /// Human-readable diagnostic.
295        message: String,
296    },
297}
298
299//--------------------------------------------------------------------------------------------------
300// Methods
301//--------------------------------------------------------------------------------------------------
302
303impl JobId {
304    /// Validate the portable opaque ID, including before it becomes a path component.
305    pub fn parse(value: impl Into<String>) -> Result<Self, String> {
306        let value = value.into();
307        if value.len() != 36
308            || !value.starts_with("job_")
309            || !value[4..]
310                .bytes()
311                .all(|b| b.is_ascii_hexdigit() && !b.is_ascii_uppercase())
312        {
313            return Err("job ID must be job_ followed by 32 lowercase hexadecimal digits".into());
314        }
315        Ok(Self(value))
316    }
317
318    /// Return the stable string representation.
319    pub fn as_str(&self) -> &str {
320        &self.0
321    }
322}
323
324impl JobState {
325    /// Whether the runtime may still own a live process for this record.
326    pub fn is_active(self) -> bool {
327        matches!(self, Self::Starting | Self::Running)
328    }
329}
330
331impl JobInfo {
332    /// Construct a pending admission without persisting the command's environment or input.
333    pub fn starting(
334        id: JobId,
335        runtime_boot_id: String,
336        command: &ExecRequest,
337        no_stdin: bool,
338        created_at: i64,
339    ) -> Self {
340        Self {
341            version: JOB_PROTOCOL_VERSION,
342            id,
343            runtime_boot_id,
344            command: std::iter::once(command.cmd.clone())
345                .chain(command.args.clone())
346                .collect(),
347            state: JobState::Starting,
348            tty: command.tty,
349            stdin_closed: no_stdin,
350            pid: None,
351            created_at,
352            started_at: None,
353            finished_at: None,
354            exit_code: None,
355            failure: None,
356            error: None,
357            timed_out: false,
358        }
359    }
360}
361
362impl JobOutputSource {
363    /// Stable textual form, identical to the serialized spelling.
364    pub fn as_str(self) -> &'static str {
365        match self {
366            Self::Stdout => "stdout",
367            Self::Stderr => "stderr",
368            Self::Output => "output",
369            Self::StdinError => "stdin_error",
370        }
371    }
372
373    /// Whether these bytes were charged against the guest's output allowance
374    /// and must be acknowledged back once consumed.
375    pub fn charged(self) -> bool {
376        !matches!(self, Self::StdinError)
377    }
378}
379
380impl std::fmt::Display for JobOutputSource {
381    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
382        f.write_str(self.as_str())
383    }
384}
385
386impl JobOutput {
387    /// Construct one captured output record.
388    pub fn new(
389        sequence: u64,
390        timestamp: i64,
391        source: JobOutputSource,
392        data_base64: String,
393    ) -> Self {
394        Self {
395            sequence,
396            timestamp,
397            source,
398            data_base64,
399        }
400    }
401}
402
403impl JobResponse {
404    /// Construct a structured failure without exposing request contents.
405    pub fn error(code: impl Into<String>, message: impl Into<String>) -> Self {
406        Self::Error {
407            code: code.into(),
408            message: message.into(),
409        }
410    }
411}
412
413impl JobPage {
414    /// Page sorted, filtered records by both count and encoded size.
415    pub fn bounded(
416        records: impl IntoIterator<Item = JobInfo>,
417        limit: usize,
418    ) -> Result<Self, String> {
419        let mut items: Vec<JobInfo> = Vec::new();
420        let mut bytes = 0;
421        let mut encoded = Vec::new();
422        for info in records {
423            encoded.clear();
424            ciborium::ser::into_writer(&info, &mut encoded).map_err(|error| error.to_string())?;
425            let size = encoded.len();
426            if limit == 0 || size > JOB_LIST_BYTES {
427                return Err("invalid page limit or oversized job metadata".into());
428            }
429            if !items.is_empty() && (items.len() >= limit || bytes + size > JOB_LIST_BYTES) {
430                return Ok(Self {
431                    next_cursor: items.last().map(|info| info.id.clone()),
432                    items,
433                });
434            }
435            bytes += size;
436            items.push(info);
437        }
438        Ok(Self {
439            items,
440            next_cursor: None,
441        })
442    }
443}
444
445//--------------------------------------------------------------------------------------------------
446// Trait Implementations
447//--------------------------------------------------------------------------------------------------
448
449impl TryFrom<String> for JobId {
450    type Error = String;
451    fn try_from(value: String) -> Result<Self, Self::Error> {
452        Self::parse(value)
453    }
454}
455impl From<JobId> for String {
456    fn from(value: JobId) -> Self {
457        value.0
458    }
459}
460impl AsRef<str> for JobId {
461    fn as_ref(&self) -> &str {
462        self.as_str()
463    }
464}
465impl std::fmt::Display for JobId {
466    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
467        f.write_str(&self.0)
468    }
469}
470
471//--------------------------------------------------------------------------------------------------
472// Tests
473//--------------------------------------------------------------------------------------------------
474
475#[cfg(test)]
476mod tests {
477    use super::*;
478
479    #[test]
480    fn job_ids_cannot_escape_their_storage_directory() {
481        for invalid in [
482            "../outside",
483            "job_../../../../../../../../../../..",
484            "job_ABCDEF",
485            "job_0000000000000000000000000000000/",
486        ] {
487            assert!(JobId::parse(invalid).is_err());
488            assert!(serde_json::from_value::<JobId>(serde_json::json!(invalid)).is_err());
489        }
490        assert!(JobId::parse("job_0123456789abcdef0123456789abcdef").is_ok());
491    }
492
493    #[test]
494    fn job_pages_are_bounded_by_encoded_bytes_as_well_as_count() {
495        let command: ExecRequest =
496            serde_json::from_value(serde_json::json!({"cmd": "x".repeat(16 * 1024)})).unwrap();
497        let records = (0..100).map(|n| {
498            JobInfo::starting(
499                JobId::parse(format!("job_{n:032x}")).unwrap(),
500                "boot".into(),
501                &command,
502                false,
503                0,
504            )
505        });
506        let page = JobPage::bounded(records, 100).unwrap();
507        assert!(page.items.len() < 100);
508        assert_eq!(
509            page.next_cursor.as_ref(),
510            page.items.last().map(|info| &info.id)
511        );
512        assert!(serde_json::to_vec(&page).unwrap().len() < JOB_REPLY_BYTES as usize);
513    }
514}
microsandbox_protocol/jobs.rs

microsandbox_protocol/
jobs.rs
