arcbox-computer-runtime 0.10.3

//! The durable phase vocabulary: what a record can say about a sandbox
//! generation, and which moves between those phases are legal.
//!
//! No filesystem and no store state — the record's own `Utc::now()` stamps
//! are the only outside input. [`super::store`] is what persists it.

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use uuid::Uuid;

use crate::error::{ComputerError, Result};
use crate::sandbox::{ComputerId, ComputerSpec, validate_id};

pub(super) const RECORD_VERSION: u32 = 1;

/// Durable control-plane phase for one sandbox generation.
///
/// Workload execution remains transient: a running workload keeps the durable
/// phase at `Ready`, avoiding record writes on the execution hot path.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ComputerPhase {
    Creating,
    Starting,
    Ready,
    Stopping,
    Stopped,
    Failed,
    Removing,
    /// Pause in progress: checkpoint taken or being taken, runtime
    /// resources not yet fully released.
    Pausing,
    /// Checkpointed with resources released; `pause_snapshot_id` names the
    /// catalog entry a Resume restores from (CORE-21).
    Paused,
    /// Resume in progress: runtime resources are being re-created.
    Resuming,
}

/// The durable phase under the name R3's lifecycle HSM uses for it: what a
/// crash-restart reads back, as opposed to the in-memory `ComputerState` a
/// caller sees.
///
/// The module boundary is where the two names meet: inside `record` the
/// enum is `ComputerPhase`, and this alias — the only one of the two
/// [`super`] re-exports — is what every other module says. R3's rename
/// then has one module to touch.
///
/// Crate-visible rather than `pub(in crate::sandbox)`: `crate::lifecycle`'s
/// state machine projects onto these phases and its table test is written
/// against [`ComputerPhase::can_transition_to`], so the durable vocabulary
/// has to reach one module outside `sandbox`.
pub type PersistPhase = ComputerPhase;

/// The stable result returned once a provisioning request has been accepted.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct ComputerProvisionOutcome {
    pub ip_address: String,
}

/// Versioned durable state for one sandbox generation.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ComputerRecord {
    pub(super) version: u32,
    pub(in crate::sandbox) id: ComputerId,
    pub(crate) generation: Uuid,
    pub(in crate::sandbox) request_key: String,
    pub(in crate::sandbox) effective_spec: ComputerSpec,
    pub(in crate::sandbox) phase: ComputerPhase,
    pub(in crate::sandbox) provision_outcome: Option<ComputerProvisionOutcome>,
    pub(in crate::sandbox) created_at: DateTime<Utc>,
    pub(in crate::sandbox) error: Option<String>,
    /// Catalog id of the internal pause checkpoint. Set while the record is
    /// `Paused`/`Resuming` so a Resume after an agent restart still finds
    /// its snapshot; cleared when the sandbox is `Ready` again.
    #[serde(default)]
    pub(in crate::sandbox) pause_snapshot_id: Option<String>,
    /// When the sandbox reached `Paused` (None otherwise).
    #[serde(default)]
    pub(in crate::sandbox) paused_at: Option<DateTime<Utc>>,
    /// When the hard maximum lifetime fires (None = no limit). Seeded from
    /// `effective_spec.ttl_seconds`; replaced by `SetLifecycle` (CORE-60).
    /// Survives `redact_runtime_inputs` — unlike the seed seconds, the
    /// deadline is durable lifecycle state, not a runtime input.
    #[serde(default)]
    pub(in crate::sandbox) ttl_deadline: Option<DateTime<Utc>>,
}

/// Result of reserving a durable provisioning intent.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ProvisionIntent {
    Created(ComputerRecord),
    Resume(ComputerRecord),
    Replay(ComputerRecord),
    Blocked(ComputerRecord),
}

/// Generation-checked lifecycle update.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ComputerTransition {
    Starting(ComputerProvisionOutcome),
    ReadyWithOutcome(ComputerProvisionOutcome),
    Ready,
    Stopping,
    Stopped,
    Failed(String),
    Removing,
    Pausing,
    Paused { snapshot_id: String },
    Resuming,
}

impl ComputerTransition {
    fn phase(&self) -> ComputerPhase {
        match self {
            Self::Starting(_) => ComputerPhase::Starting,
            Self::ReadyWithOutcome(_) | Self::Ready => ComputerPhase::Ready,
            Self::Stopping => ComputerPhase::Stopping,
            Self::Stopped => ComputerPhase::Stopped,
            Self::Failed(_) => ComputerPhase::Failed,
            Self::Removing => ComputerPhase::Removing,
            Self::Pausing => ComputerPhase::Pausing,
            Self::Paused { .. } => ComputerPhase::Paused,
            Self::Resuming => ComputerPhase::Resuming,
        }
    }
}

impl ComputerRecord {
    pub(super) fn new(id: &str, request_key: &str, effective_spec: ComputerSpec) -> Self {
        let created_at = Utc::now();
        let ttl_deadline = (effective_spec.ttl_seconds > 0)
            .then(|| created_at + chrono::Duration::seconds(i64::from(effective_spec.ttl_seconds)));

        Self {
            version: RECORD_VERSION,
            id: id.to_owned(),
            generation: Uuid::new_v4(),
            request_key: request_key.to_owned(),
            effective_spec,
            phase: ComputerPhase::Creating,
            provision_outcome: None,
            created_at,
            error: None,
            pause_snapshot_id: None,
            paused_at: None,
            ttl_deadline,
        }
    }

    pub(super) fn apply(&mut self, transition: ComputerTransition) -> Result<()> {
        let next = transition.phase();
        let atomic_ready = matches!(&transition, ComputerTransition::ReadyWithOutcome(_))
            && self.phase == ComputerPhase::Creating;
        if !atomic_ready && !self.phase.can_transition_to(next) {
            return Err(ComputerError::WrongState {
                id: self.id.clone(),
                expected: format!("a valid transition from {}", self.phase.as_str()),
                actual: next.as_str().to_owned(),
            });
        }

        match transition {
            ComputerTransition::Starting(outcome) => {
                if let Some(existing) = &self.provision_outcome
                    && existing != &outcome
                {
                    return Err(ComputerError::WrongState {
                        id: self.id.clone(),
                        expected: format!("provision outcome {existing:?}"),
                        actual: format!("provision outcome {outcome:?}"),
                    });
                }
                self.phase = ComputerPhase::Starting;
                self.provision_outcome = Some(outcome);
                self.error = None;
            }
            ComputerTransition::ReadyWithOutcome(outcome) => {
                if let Some(existing) = &self.provision_outcome
                    && existing != &outcome
                {
                    return Err(ComputerError::WrongState {
                        id: self.id.clone(),
                        expected: format!("provision outcome {existing:?}"),
                        actual: format!("provision outcome {outcome:?}"),
                    });
                }
                self.phase = ComputerPhase::Ready;
                self.provision_outcome = Some(outcome);
                self.error = None;
                self.redact_runtime_inputs();
            }
            ComputerTransition::Ready => {
                self.phase = ComputerPhase::Ready;
                self.error = None;
                self.pause_snapshot_id = None;
                self.paused_at = None;
                self.redact_runtime_inputs();
            }
            ComputerTransition::Stopping => {
                self.phase = ComputerPhase::Stopping;
                self.error = None;
                self.redact_runtime_inputs();
            }
            ComputerTransition::Stopped => {
                self.phase = ComputerPhase::Stopped;
                self.error = None;
            }
            ComputerTransition::Failed(error) => {
                self.phase = ComputerPhase::Failed;
                self.error = Some(error);
                self.redact_runtime_inputs();
            }
            ComputerTransition::Removing => {
                self.phase = ComputerPhase::Removing;
                self.redact_runtime_inputs();
            }
            ComputerTransition::Pausing => {
                self.phase = ComputerPhase::Pausing;
                self.error = None;
            }
            ComputerTransition::Paused { snapshot_id } => {
                self.phase = ComputerPhase::Paused;
                self.pause_snapshot_id = Some(snapshot_id);
                // Stamped once per pause, not once per transition: a failed
                // resume parks the record back at `Paused`, and overwriting
                // here would report the sandbox as freshly paused — and push
                // the time forward again on every retry. `Ready` clears the
                // stamp, so a genuine re-pause still gets a fresh one.
                self.paused_at.get_or_insert_with(Utc::now);
                self.error = None;
            }
            ComputerTransition::Resuming => {
                self.phase = ComputerPhase::Resuming;
                self.error = None;
            }
        }
        Ok(())
    }

    fn redact_runtime_inputs(&mut self) {
        // Checkpoints after adoption need the original boot assets.
        self.effective_spec.boot_args.clear();
        self.effective_spec.cmd.clear();
        self.effective_spec.env.clear();
        self.effective_spec.working_dir.clear();
        self.effective_spec.user.clear();
        self.effective_spec.mounts.clear();
        self.effective_spec.ttl_seconds = 0;
        self.effective_spec.ssh_public_key = None;
    }
}

impl ComputerPhase {
    pub fn can_transition_to(self, next: Self) -> bool {
        self == next
            || matches!(
                (self, next),
                (
                    Self::Creating,
                    Self::Starting | Self::Failed | Self::Removing
                ) | (
                    Self::Starting,
                    Self::Ready | Self::Stopping | Self::Failed | Self::Removing
                ) | (
                    Self::Ready,
                    Self::Stopping | Self::Pausing | Self::Failed | Self::Removing
                ) | (
                    Self::Stopping,
                    Self::Stopped | Self::Failed | Self::Removing
                ) | (Self::Stopped | Self::Failed, Self::Removing)
                    // A failed pause reverts to Ready and a completed one
                    // parks at Paused; a resume mirrors that exactly (a
                    // failed one unwinds back to Paused).
                    | (
                        Self::Pausing | Self::Resuming,
                        Self::Paused | Self::Ready | Self::Failed | Self::Removing
                    )
                    | (Self::Paused, Self::Resuming | Self::Failed | Self::Removing)
            )
    }

    pub fn as_str(self) -> &'static str {
        match self {
            Self::Creating => "creating",
            Self::Starting => "starting",
            Self::Ready => "ready",
            Self::Stopping => "stopping",
            Self::Stopped => "stopped",
            Self::Failed => "failed",
            Self::Removing => "removing",
            Self::Pausing => "pausing",
            Self::Paused => "paused",
            Self::Resuming => "resuming",
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum ExistingProvision {
    Pending,
    Replay,
    Blocked,
}

pub(super) fn classify_existing_provision(
    record: &ComputerRecord,
    request_key: &str,
) -> Result<ExistingProvision> {
    if record.request_key != request_key {
        return Err(ComputerError::AlreadyExists(record.id.clone()));
    }
    Ok(match record.phase {
        ComputerPhase::Creating => ExistingProvision::Pending,
        ComputerPhase::Starting | ComputerPhase::Ready => ExistingProvision::Replay,
        // A paused sandbox's provision outcome names a released IP, so a
        // same-key create retry must not replay it as live.
        ComputerPhase::Stopping
        | ComputerPhase::Stopped
        | ComputerPhase::Failed
        | ComputerPhase::Removing
        | ComputerPhase::Pausing
        | ComputerPhase::Paused
        | ComputerPhase::Resuming => ExistingProvision::Blocked,
    })
}

pub(super) fn validate_record(id: &str, record: &ComputerRecord) -> Result<()> {
    validate_id("sandbox id", id)?;
    if record.version != RECORD_VERSION {
        return Err(ComputerError::Config(format!(
            "unsupported sandbox record version {} for {id}",
            record.version
        )));
    }
    if record.id != id {
        return Err(ComputerError::Config(format!(
            "sandbox record id mismatch: expected {id}, got {}",
            record.id
        )));
    }
    if record.effective_spec.id.as_deref() != Some(id) {
        return Err(ComputerError::Config(format!(
            "sandbox record spec id mismatch for {id}"
        )));
    }
    if record.request_key.is_empty() {
        return Err(ComputerError::Config(format!(
            "sandbox record provision request key is empty for {id}"
        )));
    }
    if record.phase == ComputerPhase::Creating && record.provision_outcome.is_some() {
        return Err(ComputerError::Config(format!(
            "creating sandbox record unexpectedly has a provision outcome for {id}"
        )));
    }
    if matches!(
        record.phase,
        ComputerPhase::Starting
            | ComputerPhase::Ready
            | ComputerPhase::Stopping
            | ComputerPhase::Stopped
            | ComputerPhase::Pausing
            | ComputerPhase::Paused
            | ComputerPhase::Resuming
    ) && record.provision_outcome.is_none()
    {
        return Err(ComputerError::Config(format!(
            "sandbox record has no provision outcome in phase {} for {id}",
            record.phase.as_str()
        )));
    }
    if matches!(
        record.phase,
        ComputerPhase::Paused | ComputerPhase::Resuming
    ) && record.pause_snapshot_id.is_none()
    {
        return Err(ComputerError::Config(format!(
            "sandbox record has no pause snapshot in phase {} for {id}",
            record.phase.as_str()
        )));
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::ComputerPhase::*;
    use super::*;

    /// Every durable phase, in declaration order. Both axes of the edge
    /// table below iterate this, so a phase missing here is a pair nobody
    /// checks — [`targets_of`]'s exhaustive match is what stops a new
    /// variant from being added without a row.
    const ALL_PHASES: [ComputerPhase; 10] = [
        Creating, Starting, Ready, Stopping, Stopped, Failed, Removing, Pausing, Paused, Resuming,
    ];

    /// The durable edge set as a table: every phase each phase may move to,
    /// beside the self-edge all of them allow (an idempotent re-write).
    ///
    /// R3's lifecycle HSM is written against this table rather than against
    /// `can_transition_to`, which is why the test below asserts it over all
    /// `(from, to)` pairs instead of sampling: an edge missing here becomes
    /// a wrong state machine there.
    fn targets_of(from: ComputerPhase) -> &'static [ComputerPhase] {
        match from {
            Creating => &[Starting, Failed, Removing],
            Starting => &[Ready, Stopping, Failed, Removing],
            Ready => &[Stopping, Pausing, Failed, Removing],
            Stopping => &[Stopped, Failed, Removing],
            Stopped | Failed => &[Removing],
            // Terminal: `finish_remove` deletes the record rather than
            // moving it on.
            Removing => &[],
            // A failed pause reverts to Ready and a completed one parks at
            // Paused; a resume mirrors that exactly (a failed one unwinds
            // back to Paused).
            Pausing | Resuming => &[Paused, Ready, Failed, Removing],
            Paused => &[Resuming, Failed, Removing],
        }
    }

    fn outcome() -> ComputerProvisionOutcome {
        ComputerProvisionOutcome {
            ip_address: "192.0.2.2".into(),
        }
    }

    fn creating(id: &str) -> ComputerRecord {
        ComputerRecord::new(
            id,
            "key",
            ComputerSpec {
                id: Some(id.to_owned()),
                ..ComputerSpec::default()
            },
        )
    }

    #[test]
    fn the_durable_edge_set_is_exactly_this_table() {
        for from in ALL_PHASES {
            let targets = targets_of(from);
            for to in ALL_PHASES {
                let legal = from == to || targets.contains(&to);
                assert_eq!(
                    from.can_transition_to(to),
                    legal,
                    "{} -> {} should be {}",
                    from.as_str(),
                    to.as_str(),
                    if legal { "legal" } else { "refused" }
                );
            }
        }
    }

    #[test]
    fn every_transition_projects_onto_one_phase() {
        // `ComputerTransition::phase`'s own match is exhaustive, so a new
        // transition cannot skip this list without failing to compile there
        // first.
        let projections: [(ComputerTransition, ComputerPhase); 10] = [
            (ComputerTransition::Starting(outcome()), Starting),
            (ComputerTransition::ReadyWithOutcome(outcome()), Ready),
            (ComputerTransition::Ready, Ready),
            (ComputerTransition::Stopping, Stopping),
            (ComputerTransition::Stopped, Stopped),
            (ComputerTransition::Failed("boom".into()), Failed),
            (ComputerTransition::Removing, Removing),
            (ComputerTransition::Pausing, Pausing),
            (
                ComputerTransition::Paused {
                    snapshot_id: "snap".into(),
                },
                Paused,
            ),
            (ComputerTransition::Resuming, Resuming),
        ];

        for (transition, phase) in projections {
            assert_eq!(transition.phase(), phase, "{transition:?}");
        }
    }

    #[test]
    fn ready_records_keep_checkpoint_provenance_and_redact_consumed_inputs() {
        let mut record = ComputerRecord::new(
            "box",
            "key",
            ComputerSpec {
                id: Some("box".into()),
                kernel: "/assets/vmlinux".into(),
                rootfs: "/assets/rootfs.ext4".into(),
                boot_args: "console=ttyS0".into(),
                cmd: vec!["/bin/work".into()],
                env: std::collections::HashMap::from([("TOKEN".into(), "secret".into())]),
                working_dir: "/work".into(),
                user: "1000".into(),
                ttl_seconds: 60,
                ssh_public_key: Some("ssh-ed25519 key".into()),
                ..ComputerSpec::default()
            },
        );
        record
            .apply(ComputerTransition::Starting(outcome()))
            .unwrap();
        record.apply(ComputerTransition::Ready).unwrap();

        assert_eq!(record.effective_spec.kernel, "/assets/vmlinux");
        assert_eq!(record.effective_spec.rootfs, "/assets/rootfs.ext4");
        assert_eq!(record.effective_spec.boot_args, "");
        assert_eq!(record.effective_spec.cmd, Vec::<String>::new());
        assert!(record.effective_spec.env.is_empty());
        assert_eq!(record.effective_spec.working_dir, "");
        assert_eq!(record.effective_spec.user, "");
        assert_eq!(record.effective_spec.ttl_seconds, 0);
        assert_eq!(record.effective_spec.ssh_public_key, None);
        assert!(record.ttl_deadline.is_some());
    }

    #[test]
    fn only_a_restore_may_commit_ready_straight_from_creating() {
        // `Creating -> Ready` is not an edge...
        let mut record = creating("box");
        assert!(record.apply(ComputerTransition::Ready).is_err());
        assert_eq!(record.phase, Creating);

        // ...except for the restore path's single-hop commit, which carries
        // the outcome the skipped `Starting` write would have persisted.
        record
            .apply(ComputerTransition::ReadyWithOutcome(outcome()))
            .unwrap();
        assert_eq!(record.phase, Ready);
        assert_eq!(record.provision_outcome, Some(outcome()));

        // The exception is keyed on `Creating`: from anywhere else the edge
        // set rules, so a stopped record cannot be revived by it.
        record.apply(ComputerTransition::Stopping).unwrap();
        record.apply(ComputerTransition::Stopped).unwrap();
        assert!(
            record
                .apply(ComputerTransition::ReadyWithOutcome(outcome()))
                .is_err()
        );
        assert_eq!(record.phase, Stopped);
    }
}