a3s_box_core/traits/
execution.rs

1//! Backend-neutral lifecycle interface for managed A3S executions.
2
3use std::collections::BTreeMap;
4use std::num::NonZeroU16;
5use std::pin::Pin;
6use std::time::Duration;
7
8use async_trait::async_trait;
9use chrono::{DateTime, Utc};
10use serde::{Deserialize, Serialize};
11use thiserror::Error;
12use tokio::io::{AsyncRead, AsyncWrite};
13
14use crate::config::{BoxConfig, ResourceConfig};
15use crate::execution::ResolvedExecutionPlan;
16use crate::log::{LogConfig, LogEntry};
17
18/// Stable identifier assigned to one runtime execution.
19#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
20#[serde(try_from = "String", into = "String")]
21pub struct ExecutionId(String);
22
23impl ExecutionId {
24    pub fn new(value: impl Into<String>) -> ExecutionManagerResult<Self> {
25        let value = value.into();
26        if value.trim().is_empty() {
27            return Err(ExecutionManagerError::InvalidRequest(
28                "execution ID cannot be empty".to_string(),
29            ));
30        }
31        Ok(Self(value))
32    }
33
34    pub fn as_str(&self) -> &str {
35        &self.0
36    }
37}
38
39impl std::fmt::Display for ExecutionId {
40    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
41        formatter.write_str(&self.0)
42    }
43}
44
45impl TryFrom<String> for ExecutionId {
46    type Error = ExecutionManagerError;
47
48    fn try_from(value: String) -> Result<Self, Self::Error> {
49        Self::new(value)
50    }
51}
52
53impl From<ExecutionId> for String {
54    fn from(value: ExecutionId) -> Self {
55        value.0
56    }
57}
58
59/// Opaque identifier for one runtime-managed filesystem snapshot.
60///
61/// Snapshot identifiers are used as directory names below the runtime's
62/// managed snapshot root. Keeping the lexical contract here prevents callers
63/// from turning a protocol template reference into an arbitrary host path.
64#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
65#[serde(try_from = "String", into = "String")]
66pub struct ExecutionSnapshotId(String);
67
68impl ExecutionSnapshotId {
69    pub fn new(value: impl Into<String>) -> ExecutionManagerResult<Self> {
70        let value = value.into();
71        if value.is_empty()
72            || value.len() > 128
73            || !value
74                .bytes()
75                .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_'))
76        {
77            return Err(ExecutionManagerError::InvalidRequest(
78                "execution snapshot ID must match [A-Za-z0-9_-]{1,128}".to_string(),
79            ));
80        }
81        Ok(Self(value))
82    }
83
84    pub fn as_str(&self) -> &str {
85        &self.0
86    }
87}
88
89impl std::fmt::Display for ExecutionSnapshotId {
90    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
91        formatter.write_str(&self.0)
92    }
93}
94
95impl TryFrom<String> for ExecutionSnapshotId {
96    type Error = ExecutionManagerError;
97
98    fn try_from(value: String) -> Result<Self, Self::Error> {
99        Self::new(value)
100    }
101}
102
103impl From<ExecutionSnapshotId> for String {
104    fn from(value: ExecutionSnapshotId) -> Self {
105        value.0
106    }
107}
108
109/// Idempotency identity for a lifecycle operation.
110#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
111#[serde(try_from = "String", into = "String")]
112pub struct OperationId(String);
113
114impl OperationId {
115    pub fn new(value: impl Into<String>) -> ExecutionManagerResult<Self> {
116        let value = value.into();
117        if value.trim().is_empty() {
118            return Err(ExecutionManagerError::InvalidRequest(
119                "operation ID cannot be empty".to_string(),
120            ));
121        }
122        Ok(Self(value))
123    }
124
125    pub fn as_str(&self) -> &str {
126        &self.0
127    }
128}
129
130impl std::fmt::Display for OperationId {
131    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
132        formatter.write_str(&self.0)
133    }
134}
135
136impl TryFrom<String> for OperationId {
137    type Error = ExecutionManagerError;
138
139    fn try_from(value: String) -> Result<Self, Self::Error> {
140        Self::new(value)
141    }
142}
143
144impl From<OperationId> for String {
145    fn from(value: OperationId) -> Self {
146        value.0
147    }
148}
149
150/// Runtime generation used to reject stale lifecycle operations.
151#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
152#[serde(try_from = "u64", into = "u64")]
153pub struct ExecutionGeneration(u64);
154
155impl ExecutionGeneration {
156    pub const INITIAL: Self = Self(1);
157
158    pub fn new(value: u64) -> ExecutionManagerResult<Self> {
159        if value == 0 {
160            return Err(ExecutionManagerError::InvalidRequest(
161                "execution generation must be greater than zero".to_string(),
162            ));
163        }
164        Ok(Self(value))
165    }
166
167    pub const fn get(self) -> u64 {
168        self.0
169    }
170}
171
172impl TryFrom<u64> for ExecutionGeneration {
173    type Error = ExecutionManagerError;
174
175    fn try_from(value: u64) -> Result<Self, Self::Error> {
176        Self::new(value)
177    }
178}
179
180impl From<ExecutionGeneration> for u64 {
181    fn from(value: ExecutionGeneration) -> Self {
182        value.0
183    }
184}
185
186/// Restart behavior persisted with a local execution.
187#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
188#[serde(rename_all = "kebab-case")]
189pub enum ExecutionRestartPolicy {
190    /// Never restart automatically.
191    #[default]
192    No,
193    /// Restart after every exit.
194    Always,
195    /// Restart only after an unsuccessful exit.
196    OnFailure,
197    /// Restart unless a user explicitly stopped the execution.
198    UnlessStopped,
199}
200
201impl ExecutionRestartPolicy {
202    /// Canonical value stored in the backwards-compatible local record.
203    pub const fn as_str(self) -> &'static str {
204        match self {
205            Self::No => "no",
206            Self::Always => "always",
207            Self::OnFailure => "on-failure",
208            Self::UnlessStopped => "unless-stopped",
209        }
210    }
211}
212
213/// Health-check behavior persisted with a local execution.
214#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
215pub struct ExecutionHealthCheck {
216    /// Command executed by the health check.
217    pub cmd: Vec<String>,
218    /// Interval between checks in seconds.
219    #[serde(default = "default_health_interval")]
220    pub interval_secs: u64,
221    /// Per-check timeout in seconds.
222    #[serde(default = "default_health_timeout")]
223    pub timeout_secs: u64,
224    /// Consecutive failures before the execution is unhealthy.
225    #[serde(default = "default_health_retries")]
226    pub retries: u32,
227    /// Grace period after startup in seconds.
228    #[serde(default)]
229    pub start_period_secs: u64,
230}
231
232/// Caller-owned policy projected into the canonical local execution record.
233///
234/// The complete value is persisted with the creation request so retries cannot
235/// silently reuse an execution with different lifecycle or local resource
236/// policy. Runtime launch requirements remain in [`BoxConfig`].
237#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
238pub struct ExecutionRecordPolicy {
239    /// User-visible local name. `None` lets the runtime assign a safe name.
240    #[serde(default)]
241    pub name: Option<String>,
242    /// Remove the execution automatically after it stops.
243    #[serde(default)]
244    pub auto_remove: bool,
245    /// Automatic restart behavior.
246    #[serde(default)]
247    pub restart_policy: ExecutionRestartPolicy,
248    /// Maximum automatic restart count, where zero means unlimited.
249    #[serde(default)]
250    pub max_restart_count: u32,
251    /// Effective caller or cached-image health check.
252    #[serde(default)]
253    pub health_check: Option<ExecutionHealthCheck>,
254    /// Prevent a later image-defined health check from being enabled.
255    #[serde(default)]
256    pub healthcheck_disabled: bool,
257    /// Runtime log driver policy.
258    #[serde(default)]
259    pub log_config: LogConfig,
260    /// Named volumes represented by the resolved mounts in [`BoxConfig`].
261    #[serde(default)]
262    pub volume_names: Vec<String>,
263    /// Requested OCI platform retained for inspection and image selection.
264    #[serde(default)]
265    pub platform: Option<String>,
266    /// Whether the caller requested an init process.
267    #[serde(default)]
268    pub init: bool,
269    /// Requested host device mappings.
270    #[serde(default)]
271    pub devices: Vec<String>,
272    /// Requested GPU selection.
273    #[serde(default)]
274    pub gpus: Option<String>,
275    /// Shared-memory size in bytes.
276    #[serde(default)]
277    pub shm_size: Option<u64>,
278    /// Signal used for graceful stop.
279    #[serde(default)]
280    pub stop_signal: Option<String>,
281    /// Graceful stop timeout in seconds.
282    #[serde(default)]
283    pub stop_timeout: Option<u64>,
284    /// Whether the caller requested OOM-killer suppression.
285    #[serde(default)]
286    pub oom_kill_disable: bool,
287    /// Requested host OOM score adjustment.
288    #[serde(default)]
289    pub oom_score_adj: Option<i32>,
290}
291
292impl Default for ExecutionRecordPolicy {
293    fn default() -> Self {
294        Self {
295            name: None,
296            auto_remove: false,
297            restart_policy: ExecutionRestartPolicy::No,
298            max_restart_count: 0,
299            health_check: None,
300            healthcheck_disabled: false,
301            log_config: LogConfig::default(),
302            volume_names: Vec::new(),
303            platform: None,
304            init: false,
305            devices: Vec::new(),
306            gpus: None,
307            shm_size: None,
308            stop_signal: None,
309            stop_timeout: None,
310            oom_kill_disable: false,
311            oom_score_adj: None,
312        }
313    }
314}
315
316fn default_health_interval() -> u64 {
317    30
318}
319
320fn default_health_timeout() -> u64 {
321    5
322}
323
324fn default_health_retries() -> u32 {
325    3
326}
327
328/// A fully resolved request submitted to the runtime lifecycle facade.
329#[derive(Debug, Clone, Serialize, Deserialize)]
330pub struct CreateExecutionRequest {
331    /// Public identity used only as an untrusted diagnostic label.
332    pub external_sandbox_id: String,
333    /// Backend-neutral runtime configuration resolved from template policy.
334    pub config: BoxConfig,
335    /// Labels persisted with the internal execution.
336    pub labels: BTreeMap<String, String>,
337    /// Caller-owned lifecycle and local record policy.
338    #[serde(default)]
339    pub policy: ExecutionRecordPolicy,
340    /// Runtime-managed filesystem snapshot used as this execution's immutable
341    /// rootfs lower. The runtime derives the host path from this validated ID;
342    /// callers never supply a host path.
343    #[serde(default)]
344    pub rootfs_snapshot_id: Option<ExecutionSnapshotId>,
345}
346
347/// Durable evidence returned after an execution is created but not started.
348#[derive(Debug, Clone, Serialize, Deserialize)]
349pub struct ExecutionReservation {
350    pub execution_id: ExecutionId,
351    pub generation: ExecutionGeneration,
352    pub plan: ResolvedExecutionPlan,
353    pub resources: ResourceConfig,
354    pub created_at: DateTime<Utc>,
355}
356
357/// Evidence returned when a runtime execution is ready.
358#[derive(Debug, Clone, Serialize, Deserialize)]
359pub struct ExecutionLease {
360    pub execution_id: ExecutionId,
361    pub generation: ExecutionGeneration,
362    pub plan: ResolvedExecutionPlan,
363    pub resources: ResourceConfig,
364    pub started_at: DateTime<Utc>,
365}
366
367/// Result of atomically capturing one execution filesystem.
368#[derive(Debug, Clone, Serialize, Deserialize)]
369pub struct ExecutionSnapshot {
370    pub snapshot_id: ExecutionSnapshotId,
371    pub size_bytes: u64,
372    /// Stable state restored after the temporary snapshot pause.
373    pub state: ExecutionState,
374    /// Generation-fenced runtime evidence after snapshot completion.
375    pub lease: ExecutionLease,
376}
377
378/// Runtime state visible through the backend-neutral lifecycle facade.
379#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
380#[serde(rename_all = "snake_case")]
381pub enum ExecutionState {
382    Created,
383    Creating,
384    Running,
385    Paused,
386    Stopped,
387    Failed,
388}
389
390/// Current state and generation of one execution.
391#[derive(Debug, Clone, Serialize, Deserialize)]
392pub struct ExecutionStatus {
393    pub execution_id: ExecutionId,
394    pub generation: ExecutionGeneration,
395    pub state: ExecutionState,
396    pub plan: ResolvedExecutionPlan,
397}
398
399/// Result of an idempotent runtime kill request.
400#[derive(Debug, Clone, Copy, PartialEq, Eq)]
401pub enum KillOutcome {
402    Killed,
403    AlreadyStopped,
404}
405
406/// Per-operation controls persisted with an idempotent restart.
407#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
408pub struct RestartExecutionOptions {
409    /// Graceful stop deadline for the old runtime. `None` uses persisted
410    /// execution policy or the backend default.
411    #[serde(default)]
412    pub stop_timeout_secs: Option<u64>,
413}
414
415/// Runtime evidence recovered after a service restart.
416#[derive(Debug, Clone)]
417pub enum ReconcileOutcome {
418    Absent,
419    Created(ExecutionReservation),
420    Creating,
421    Ready(ExecutionLease),
422    Failed,
423}
424
425/// Errors returned by the lifecycle facade without exposing backend internals.
426#[derive(Debug, Error)]
427pub enum ExecutionManagerError {
428    #[error("invalid execution request: {0}")]
429    InvalidRequest(String),
430    #[error("execution not found: {0}")]
431    NotFound(ExecutionId),
432    #[error("execution conflict for {execution_id}: {message}")]
433    Conflict {
434        execution_id: ExecutionId,
435        message: String,
436    },
437    #[error("execution backend unavailable: {0}")]
438    Unavailable(String),
439    #[error("execution lifecycle failed: {0}")]
440    Internal(String),
441}
442
443pub type ExecutionManagerResult<T> = std::result::Result<T, ExecutionManagerError>;
444
445/// Bidirectional byte stream connected to one generation-fenced workload port.
446pub trait ExecutionPortIo: AsyncRead + AsyncWrite + Send + Unpin {}
447
448impl<T> ExecutionPortIo for T where T: AsyncRead + AsyncWrite + Send + Unpin {}
449
450pub type ExecutionPortStream = Pin<Box<dyn ExecutionPortIo>>;
451
452/// Backend-neutral connector used by data-plane gateways.
453///
454/// Implementations must validate the execution generation atomically with
455/// selecting the live runtime. A connector must never fall back to another
456/// execution or generation when the requested runtime is unavailable.
457#[async_trait]
458pub trait ExecutionPortConnector: Send + Sync {
459    async fn connect_port(
460        &self,
461        execution_id: &ExecutionId,
462        generation: ExecutionGeneration,
463        port: NonZeroU16,
464        timeout: Duration,
465    ) -> ExecutionManagerResult<ExecutionPortStream>;
466}
467
468/// Backend-neutral lifecycle facade shared by the CLI, SDK, and remote service.
469#[async_trait]
470pub trait ExecutionManager: Send + Sync {
471    /// Persist exactly one unstarted execution reservation for `operation_id`.
472    async fn create(
473        &self,
474        _request: CreateExecutionRequest,
475        _operation_id: &OperationId,
476    ) -> ExecutionManagerResult<ExecutionReservation> {
477        Err(ExecutionManagerError::Unavailable(
478            "this execution manager does not support staged create".to_string(),
479        ))
480    }
481
482    /// Start one created execution after fencing stale callers by generation.
483    async fn start(
484        &self,
485        _execution_id: &ExecutionId,
486        _generation: ExecutionGeneration,
487    ) -> ExecutionManagerResult<ExecutionLease> {
488        Err(ExecutionManagerError::Unavailable(
489            "this execution manager does not support staged start".to_string(),
490        ))
491    }
492
493    /// Create and start exactly one execution for `operation_id`.
494    ///
495    /// Retrying after a crash reuses the durable reservation and continues its
496    /// start instead of allocating a second execution.
497    async fn create_and_start(
498        &self,
499        request: CreateExecutionRequest,
500        operation_id: &OperationId,
501    ) -> ExecutionManagerResult<ExecutionLease> {
502        let reservation = self.create(request, operation_id).await?;
503        self.start(&reservation.execution_id, reservation.generation)
504            .await
505    }
506
507    async fn inspect(&self, execution_id: &ExecutionId) -> ExecutionManagerResult<ExecutionStatus>;
508
509    /// Read structured stdout/stderr entries after fencing the runtime generation.
510    async fn read_logs(
511        &self,
512        _execution_id: &ExecutionId,
513        _generation: ExecutionGeneration,
514    ) -> ExecutionManagerResult<Vec<LogEntry>> {
515        Err(ExecutionManagerError::Unavailable(
516            "this execution manager does not expose structured logs".to_string(),
517        ))
518    }
519
520    /// Temporarily quiesce the execution, atomically capture its rootfs in the
521    /// runtime-managed snapshot store, and restore its prior stable state.
522    async fn create_filesystem_snapshot(
523        &self,
524        _execution_id: &ExecutionId,
525        _generation: ExecutionGeneration,
526        _snapshot_id: &ExecutionSnapshotId,
527    ) -> ExecutionManagerResult<ExecutionSnapshot> {
528        Err(ExecutionManagerError::Unavailable(
529            "this execution manager does not support filesystem snapshots".to_string(),
530        ))
531    }
532
533    /// Return the size of a fully published runtime-managed snapshot, or
534    /// `None` when it does not exist.
535    async fn filesystem_snapshot_size(
536        &self,
537        _snapshot_id: &ExecutionSnapshotId,
538    ) -> ExecutionManagerResult<Option<u64>> {
539        Err(ExecutionManagerError::Unavailable(
540            "this execution manager does not expose filesystem snapshots".to_string(),
541        ))
542    }
543
544    /// Delete a runtime-managed snapshot, refusing while an active execution
545    /// still uses it as a copy-on-write lower.
546    async fn delete_filesystem_snapshot(
547        &self,
548        _snapshot_id: &ExecutionSnapshotId,
549    ) -> ExecutionManagerResult<bool> {
550        Err(ExecutionManagerError::Unavailable(
551            "this execution manager does not support filesystem snapshot deletion".to_string(),
552        ))
553    }
554
555    /// Pause one execution and return the generation-fenced paused lease.
556    async fn pause(
557        &self,
558        execution_id: &ExecutionId,
559        generation: ExecutionGeneration,
560        keep_memory: bool,
561    ) -> ExecutionManagerResult<ExecutionLease>;
562
563    async fn resume(
564        &self,
565        execution_id: &ExecutionId,
566        generation: ExecutionGeneration,
567    ) -> ExecutionManagerResult<ExecutionLease>;
568
569    /// Terminate the current runtime, advance its generation exactly once,
570    /// and start it again under an idempotent operation identity.
571    async fn restart(
572        &self,
573        execution_id: &ExecutionId,
574        generation: ExecutionGeneration,
575        operation_id: &OperationId,
576    ) -> ExecutionManagerResult<ExecutionLease> {
577        self.restart_with_options(
578            execution_id,
579            generation,
580            operation_id,
581            RestartExecutionOptions::default(),
582        )
583        .await
584    }
585
586    /// Restart with controls that become part of the durable operation intent.
587    async fn restart_with_options(
588        &self,
589        _execution_id: &ExecutionId,
590        _generation: ExecutionGeneration,
591        _operation_id: &OperationId,
592        _options: RestartExecutionOptions,
593    ) -> ExecutionManagerResult<ExecutionLease> {
594        Err(ExecutionManagerError::Unavailable(
595            "this execution manager does not support restart".to_string(),
596        ))
597    }
598
599    async fn kill(
600        &self,
601        execution_id: &ExecutionId,
602        generation: ExecutionGeneration,
603    ) -> ExecutionManagerResult<KillOutcome>;
604
605    async fn reconcile(
606        &self,
607        operation_id: &OperationId,
608    ) -> ExecutionManagerResult<ReconcileOutcome>;
609}
610
611#[cfg(test)]
612mod tests {
613    use super::*;
614
615    #[test]
616    fn identifiers_reject_empty_values() {
617        assert!(matches!(
618            ExecutionId::new("  "),
619            Err(ExecutionManagerError::InvalidRequest(_))
620        ));
621        assert!(matches!(
622            OperationId::new(""),
623            Err(ExecutionManagerError::InvalidRequest(_))
624        ));
625    }
626
627    #[test]
628    fn generation_rejects_zero() {
629        assert!(matches!(
630            ExecutionGeneration::new(0),
631            Err(ExecutionManagerError::InvalidRequest(_))
632        ));
633        assert_eq!(ExecutionGeneration::INITIAL.get(), 1);
634        assert!(serde_json::from_str::<ExecutionGeneration>("0").is_err());
635    }
636
637    #[test]
638    fn identifier_deserialization_preserves_invariants() {
639        assert!(serde_json::from_str::<ExecutionId>("\"\"").is_err());
640        assert!(serde_json::from_str::<OperationId>("\" \"").is_err());
641    }
642
643    #[test]
644    fn snapshot_identifiers_are_safe_managed_directory_names() {
645        for valid in ["snapshot-1", "SNAPSHOT_2", "a"] {
646            assert_eq!(ExecutionSnapshotId::new(valid).unwrap().as_str(), valid);
647        }
648        for invalid in [
649            "",
650            ".",
651            "..",
652            "../snapshot",
653            "snapshot/path",
654            "snapshot:tag",
655            "snapshot id",
656        ] {
657            assert!(matches!(
658                ExecutionSnapshotId::new(invalid),
659                Err(ExecutionManagerError::InvalidRequest(_))
660            ));
661        }
662        assert!(ExecutionSnapshotId::new("x".repeat(129)).is_err());
663        assert!(serde_json::from_str::<ExecutionSnapshotId>("\"../snapshot\"").is_err());
664    }
665
666    #[test]
667    fn legacy_creation_requests_default_record_policy() {
668        let request: CreateExecutionRequest = serde_json::from_value(serde_json::json!({
669            "external_sandbox_id": "sandbox-1",
670            "config": BoxConfig::default(),
671            "labels": {"purpose": "compatibility"}
672        }))
673        .unwrap();
674
675        assert_eq!(request.policy, ExecutionRecordPolicy::default());
676        assert_eq!(request.policy.restart_policy, ExecutionRestartPolicy::No);
677        assert!(request.rootfs_snapshot_id.is_none());
678    }
679
680    #[test]
681    fn restart_policy_has_stable_record_values() {
682        assert_eq!(ExecutionRestartPolicy::No.as_str(), "no");
683        assert_eq!(ExecutionRestartPolicy::Always.as_str(), "always");
684        assert_eq!(ExecutionRestartPolicy::OnFailure.as_str(), "on-failure");
685        assert_eq!(
686            ExecutionRestartPolicy::UnlessStopped.as_str(),
687            "unless-stopped"
688        );
689        assert_eq!(
690            serde_json::to_value(ExecutionRestartPolicy::OnFailure).unwrap(),
691            "on-failure"
692        );
693    }
694}
a3s_box_core/traits/execution.rs

a3s_box_core/traits/
execution.rs
