artifactize_core/runs/
mod.rs

1//! Artifactize-owned verification: durable submission, observation and explicit cancellation.
2#[cfg(test)]
3mod lifetime_tests;
4mod observe;
5mod result;
6mod worker;
7use crate::{
8    process,
9    project::{VerifyOptions, selection::Selection},
10    store::{self, Receipts},
11    types::{FailureCode, RequestKey, RunId, RunStatus},
12};
13pub use observe::{Counts, RunObservation, State, status, view, wait};
14pub use result::{Finding, ResultCursor, ResultState, RunResultPage, RunVerdict, result};
15use serde::{Deserialize, Serialize};
16use std::path::{Path, PathBuf};
17pub(crate) use worker::{execute, supervise};
18
19/// Per-Run execution modes. Future modes extend this type, not transport signatures.
20#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
21#[serde(default)]
22pub struct RunModes {
23    pub speculate: bool,
24    pub defer_human: bool,
25}
26impl RunModes {
27    pub(crate) fn is_default(&self) -> bool {
28        self == &Self::default()
29    }
30}
31
32#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
33#[serde(rename_all = "snake_case")]
34pub enum Activity {
35    Preparing,
36    Executing,
37    Publishing,
38    Waiting,
39    Cancelling,
40}
41
42#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
43#[serde(tag = "state", rename_all = "snake_case")]
44pub enum Worker {
45    Starting {
46        launcher: process::ChildIdentity,
47    },
48    Admitted {
49        owner: process::ChildIdentity,
50    },
51    /// Old records have no worker identity and are not evidence of owner death.
52    Legacy,
53}
54#[derive(Debug, Clone, Serialize, Deserialize)]
55#[serde(rename_all = "camelCase")]
56pub struct Control {
57    pub worker: Worker,
58    /// The supervisor keeps worker files open until its executor has exited and been checked.
59    #[serde(default, skip_serializing_if = "Option::is_none")]
60    pub supervisor: Option<process::ChildIdentity>,
61    pub activity: Activity,
62    #[serde(default)]
63    pub cancellation_requested: bool,
64    #[serde(default, skip_serializing_if = "Option::is_none")]
65    pub failure_code: Option<FailureCode>,
66}
67impl Control {
68    pub(crate) fn starting() -> Result<Self, String> {
69        Ok(Self {
70            worker: Worker::Starting {
71                launcher: process::child_identity(std::process::id()).map_err(|e| e.to_string())?,
72            },
73            supervisor: None,
74            activity: Activity::Preparing,
75            cancellation_requested: false,
76            failure_code: None,
77        })
78    }
79    pub(crate) fn legacy() -> Self {
80        Self {
81            worker: Worker::Legacy,
82            supervisor: None,
83            activity: Activity::Executing,
84            cancellation_requested: false,
85            failure_code: None,
86        }
87    }
88}
89
90/// Bound repository/state and trusted executable, resolved once by the interface at startup.
91#[derive(Debug, Clone)]
92pub struct Context {
93    pub(crate) repo: PathBuf,
94    pub(crate) state: PathBuf,
95    executable: PathBuf,
96}
97impl Context {
98    pub fn new(repo: &Path, state: Option<&Path>, executable: &Path) -> Result<Self, String> {
99        Ok(Self {
100            repo: crate::workspace::canonical_target(repo).map_err(|e| e.to_string())?,
101            state: crate::workspace::canonical_target(&store::state_dir(state)?)
102                .map_err(|e| e.to_string())?,
103            executable: executable.to_path_buf(),
104        })
105    }
106    /// Public-library convenience; packaged callers normally pass their own executable to `new`.
107    pub fn discover(repo: &Path, state: Option<&Path>) -> Result<Self, String> {
108        let executable = process::artifactize_executable()?;
109        Self::new(repo, state, &executable)
110    }
111}
112
113pub async fn start(
114    context: &Context,
115    selection: &Selection,
116    options: &VerifyOptions,
117    request_key: Option<RequestKey>,
118) -> Result<RunObservation, String> {
119    // Keys are attempt identity, not checkout identity. Retry before discovering declarations.
120    if let Some(key) = &request_key
121        && context.state.join(store::DATABASE).is_file()
122    {
123        let receipts = Receipts::open(&context.state, &context.repo).await?;
124        let result = receipts.keyed_run(&context.repo, key).await;
125        let closed = receipts.close().await;
126        if let Some(id) = result.and_then(|id| closed.map(|()| id))? {
127            return status(context, &id).await;
128        }
129    }
130    let (id, accepted) = crate::project::verify::reserve(
131        &context.repo,
132        Some(&context.state),
133        selection,
134        options,
135        request_key.as_ref(),
136    )
137    .await?;
138    if accepted {
139        let receipts = Receipts::open(&context.state, &context.repo).await?;
140        let result = async {
141            let launch =
142                process::launch_run(&context.executable, &id, &context.repo, &context.state);
143            let admitted = match launch {
144                Ok(identity) => receipts.admit(&id, identity).await,
145                Err(error) => Err(error.to_string()),
146            };
147            if let Err(error) = admitted {
148                worker::fail(&receipts, &id, FailureCode::SpawnFailed, error.clone()).await?;
149                return Err(format!("Run {id} handoff failed: {error}"));
150            }
151            Ok(())
152        }
153        .await;
154        let closed = receipts.close().await;
155        result.and(closed)?;
156    }
157    status(context, &id).await
158}
159
160#[derive(Debug, Clone, Serialize)]
161pub struct CancelAcknowledgement {
162    pub run: RunId,
163    pub state: State,
164    pub cancellation_requested: bool,
165}
166pub async fn cancel(context: &Context, id: &RunId) -> Result<CancelAcknowledgement, String> {
167    let current = view(context, id).await?;
168    let receipts = Receipts::open(&context.state, &context.repo).await?;
169    let result = receipts.request_cancel(&current.run.id).await;
170    let closed = receipts.close().await;
171    let run = result.and_then(|run| closed.map(|()| run))?;
172    Ok(CancelAcknowledgement {
173        run: id.clone(),
174        state: observe::state(&run, &current.requests),
175        cancellation_requested: run
176            .control
177            .as_ref()
178            .is_some_and(|control| control.cancellation_requested),
179    })
180}
181
182pub(crate) async fn reconcile(
183    state: &Path,
184    view: store::RunView,
185) -> Result<store::RunView, String> {
186    if view.run.status() == RunStatus::Running
187        && let Some(control) = &view.run.control
188    {
189        let owner = match control.worker {
190            Worker::Starting { launcher } => Some(launcher),
191            Worker::Admitted { owner } => Some(owner),
192            Worker::Legacy => None,
193        };
194        if owner.is_some() && !process::is_alive(owner).map_err(|e| e.to_string())? {
195            worker::recover(state, &view).await?;
196            return store::read_run(state, &view.run.id).await;
197        }
198    }
199    Ok(view)
200}
artifactize_core/runs/mod.rs

artifactize_core/runs/
mod.rs
