optimization_job_api/
artifact.rs

1//! Artifacts: files of a job attempt that agents may read through the runner
2//! (`jobs artifact`, MCP `get_artifact`), during the run and after it.
3//!
4//! `results` (the result journal) is built in. A worker adds its own files,
5//! such as an evaluation log or a warm-start snapshot, by declaring them with
6//! [`declare`] before it first writes them. Declarations are checked records
7//! in `artifacts.jsonl` of the attempt directory, so workers in any language
8//! can write them, and a runner that predates them simply ignores the file.
9//! The final manifest lists every declared file that exists, and
10//! `runner.artifact_index` records its format, size and SHA-256 when the
11//! attempt is sealed.
12use crate::{Result, coded, safe_name, storage};
13use serde::{Deserialize, Serialize};
14use serde_json::{Map, Value, json};
15use std::{
16    collections::BTreeSet,
17    fs,
18    io::{self, BufRead, Read},
19    path::Path,
20};
21
22/// The declarations of an attempt: checked records, version 1.
23pub const DECLARATIONS: &str = "artifacts.jsonl";
24const LOCK: &str = "artifacts.lock";
25/// The built-in artifact: the result journal `results.jsonl`.
26pub const RESULTS: &str = "results";
27/// Artifact formats.
28///
29/// - `records`: an append-only journal of checked records, written with
30///   [`storage::append_checked`] (any record version). Readable while the
31///   job runs (complete records only) and in pages of records.
32/// - `json`: one JSON document, replaced atomically with
33///   [`storage::write_atomic_json`]. Read whole.
34pub const FORMATS: [&str; 2] = ["records", "json"];
35/// Files of the runner and the SDK in an attempt directory. No artifact may
36/// use them; nor a name that starts with `.` or ends with `.lock` or `.tmp`.
37pub const RESERVED: [&str; 22] = [
38    "artifacts.jsonl",
39    "artifacts.lock",
40    "cancel.lock",
41    "cancel.request",
42    "child_finished",
43    "containment.unknown",
44    "events.jsonl",
45    "inputs",
46    "job.json",
47    "job.lock",
48    "provenance.json",
49    "request.json",
50    "resolved.json",
51    "results.jsonl",
52    "retries.jsonl",
53    "run.json",
54    "stderr.log",
55    "stdout.log",
56    "submitted.json",
57    "worker",
58    "worker-process.json",
59    "worker.lock",
60];
61/// Declarations per attempt.
62pub const MAX_DECLARATIONS: usize = 4096;
63/// Bytes of a description.
64pub const MAX_DESCRIPTION: usize = 200;
65/// What one runner call returns inline: the whole content of an artifact,
66/// or the records of one page.
67pub const INLINE_CAP: usize = 1 << 20;
68/// Records of one page at most.
69pub const MAX_PAGE: usize = 10_000;
70
71/// A file a worker declares in its attempt directory.
72#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
73#[serde(deny_unknown_fields)]
74pub struct Declaration {
75    /// Name agents use (`jobs artifact ID NAME`); unique in the attempt.
76    pub name: String,
77    /// File name in the attempt directory, no path.
78    pub file: String,
79    /// One of [`FORMATS`].
80    pub format: String,
81    /// The task that writes it, if one does.
82    #[serde(default, skip_serializing_if = "Option::is_none")]
83    pub task_id: Option<usize>,
84    /// What it holds and what it is for, for agents.
85    #[serde(default, skip_serializing_if = "String::is_empty")]
86    pub description: String,
87}
88
89impl Declaration {
90    /// A journal of checked records (see [`FORMATS`]).
91    pub fn records(name: impl Into<String>, file: impl Into<String>) -> Self {
92        Self {
93            name: name.into(),
94            file: file.into(),
95            format: "records".into(),
96            task_id: None,
97            description: String::new(),
98        }
99    }
100    /// A JSON document replaced atomically (see [`FORMATS`]).
101    pub fn json(name: impl Into<String>, file: impl Into<String>) -> Self {
102        Self {
103            format: "json".into(),
104            ..Self::records(name, file)
105        }
106    }
107    pub fn task(mut self, task_id: usize) -> Self {
108        self.task_id = Some(task_id);
109        self
110    }
111    pub fn describe(mut self, description: impl Into<String>) -> Self {
112        self.description = description.into();
113        self
114    }
115    /// The rules of a declaration, independent of other declarations.
116    pub fn check(&self) -> std::result::Result<(), String> {
117        let file = self.file.as_str();
118        if !safe_name(&self.name) || self.name == RESULTS {
119            return Err(format!("invalid or reserved artifact name {:?}", self.name));
120        }
121        if !safe_name(file)
122            || RESERVED.contains(&file)
123            || file.starts_with('.')
124            || file.ends_with(".lock")
125            || file.ends_with(".tmp")
126        {
127            return Err(format!(
128                "artifact {}: invalid or reserved file name {file:?}",
129                self.name
130            ));
131        }
132        if !FORMATS.contains(&self.format.as_str()) {
133            return Err(format!(
134                "artifact {}: unknown format {:?}",
135                self.name, self.format
136            ));
137        }
138        if self.description.len() > MAX_DESCRIPTION || self.description.contains(char::is_control) {
139            return Err(format!(
140                "artifact {}: description over {MAX_DESCRIPTION} bytes or with control characters",
141                self.name
142            ));
143        }
144        Ok(())
145    }
146}
147
148/// Declares an artifact of the attempt in `dir`, the attempt directory a
149/// worker's `prepare` receives as `base`. Declare it in `solve`, before the
150/// first write, so agents can read it while the job runs; never in `prepare`,
151/// which also runs for validation. Repeating an identical declaration is a
152/// no-op; another declaration of the same name or file is an error. Safe to
153/// call from concurrent tasks.
154pub fn declare(dir: &Path, declaration: &Declaration) -> Result<()> {
155    declaration.check()?;
156    let _lock = storage::lock(&dir.join(LOCK))?;
157    let existing = declarations(dir)?;
158    if let Some(old) = existing.iter().find(|d| d.name == declaration.name) {
159        if old == declaration {
160            return Ok(());
161        }
162        return Err(format!(
163            "artifact {} is already declared differently",
164            declaration.name
165        )
166        .into());
167    }
168    if existing.iter().any(|d| d.file == declaration.file) {
169        return Err(format!("file {} is already declared", declaration.file).into());
170    }
171    if existing.len() >= MAX_DECLARATIONS {
172        return Err(format!("more than {MAX_DECLARATIONS} artifacts declared").into());
173    }
174    storage::append_checked(&dir.join(DECLARATIONS), 1, declaration)?;
175    Ok(())
176}
177
178/// The declarations of the attempt in `dir`, in declaration order; none if
179/// there are none. A declaration that breaks the rules, or a second
180/// declaration of a name or file, makes them all untrusted: an error.
181pub fn declarations(dir: &Path) -> Result<Vec<Declaration>> {
182    let path = dir.join(DECLARATIONS);
183    if !path.exists() {
184        return Ok(vec![]);
185    }
186    let list: Vec<Declaration> =
187        storage::read_checked(&path, 1).map_err(|e| format!("{DECLARATIONS}: {e}"))?;
188    let (mut names, mut files) = (BTreeSet::new(), BTreeSet::new());
189    for d in &list {
190        d.check().map_err(|e| format!("{DECLARATIONS}: {e}"))?;
191        if !names.insert(&d.name) || !files.insert(&d.file) {
192            return Err(format!("{DECLARATIONS}: {} or its file declared twice", d.name).into());
193        }
194    }
195    if list.len() > MAX_DECLARATIONS {
196        return Err(format!("{DECLARATIONS}: more than {MAX_DECLARATIONS} artifacts").into());
197    }
198    Ok(list)
199}
200
201/// Every artifact of the attempt: `results` first, then the declarations.
202/// Files that do not exist (yet) are included.
203pub fn all(dir: &Path) -> Result<Vec<Declaration>> {
204    let mut list = vec![results()];
205    list.extend(declarations(dir)?);
206    Ok(list)
207}
208
209fn results() -> Declaration {
210    Declaration::records(RESULTS, "results.jsonl")
211        .describe("The result journal: one checked record per completed task")
212}
213
214/// [`all`], or `results` alone and the error when the declarations are
215/// broken: what a running attempt serves.
216pub fn all_or_results(dir: &Path) -> (Vec<Declaration>, Vec<String>) {
217    match all(dir) {
218        Ok(list) => (list, vec![]),
219        Err(e) => (vec![results()], vec![e.to_string()]),
220    }
221}
222
223fn entry(d: &Declaration) -> Map<String, Value> {
224    let mut e = Map::new();
225    e.insert("file".into(), json!(d.file));
226    e.insert("format".into(), json!(d.format));
227    if let Some(task) = d.task_id {
228        e.insert("task_id".into(), json!(task));
229    }
230    if !d.description.is_empty() {
231        e.insert("description".into(), json!(d.description));
232    }
233    e
234}
235
236/// The artifact index of a running attempt, for `jobs get`: every artifact
237/// whose file exists, with its current size, and the problems found (a broken
238/// declaration file leaves `results` alone).
239pub fn progress(dir: &Path) -> (Value, Vec<String>) {
240    let (list, errors) = all_or_results(dir);
241    let mut index = Map::new();
242    for d in list {
243        if let Ok(meta) = fs::symlink_metadata(dir.join(&d.file))
244            && meta.is_file()
245        {
246            let mut e = entry(&d);
247            e.insert("bytes".into(), json!(meta.len()));
248            index.insert(d.name.clone(), Value::Object(e));
249        }
250    }
251    (Value::Object(index), errors)
252}
253
254/// The manifest's view of the artifacts when an attempt is sealed: the
255/// `artifacts` map (name to file), `runner.artifact_index` (format, size,
256/// SHA-256, task, description) and the problems found. Only regular files
257/// that exist are listed; a broken declaration file lists `results` alone.
258/// An artifact problem never fails the attempt: its results stand.
259pub fn sealed(dir: &Path) -> (Value, Value, Vec<String>) {
260    let (list, mut errors) = all_or_results(dir);
261    let (mut files, mut index) = (Map::new(), Map::new());
262    for d in list {
263        let path = dir.join(&d.file);
264        match fs::symlink_metadata(&path) {
265            Ok(meta) if meta.is_file() => {}
266            Err(e) if e.kind() == io::ErrorKind::NotFound => continue,
267            Ok(_) => {
268                errors.push(format!(
269                    "artifact {}: {} is not a regular file",
270                    d.name, d.file
271                ));
272                continue;
273            }
274            Err(e) => {
275                errors.push(format!("artifact {}: {e}", d.name));
276                continue;
277            }
278        }
279        let (bytes, sha256) = match digest_file(&path) {
280            Ok(digest) => digest,
281            Err(e) => {
282                errors.push(format!("artifact {}: {e}", d.name));
283                continue;
284            }
285        };
286        let mut e = entry(&d);
287        e.insert("bytes".into(), json!(bytes));
288        e.insert("sha256".into(), json!(sha256));
289        files.insert(d.name.clone(), json!(d.file));
290        index.insert(d.name, Value::Object(e));
291    }
292    (Value::Object(files), Value::Object(index), errors)
293}
294
295/// Size and SHA-256 of a file, read in blocks.
296pub fn digest_file(path: &Path) -> Result<(u64, String)> {
297    use sha2::{Digest, Sha256};
298    let mut file = fs::File::open(path)?;
299    let mut hasher = Sha256::new();
300    let mut buffer = vec![0u8; 1 << 16];
301    let mut bytes = 0u64;
302    loop {
303        let n = file.read(&mut buffer)?;
304        if n == 0 {
305            break;
306        }
307        bytes += n as u64;
308        hasher.update(&buffer[..n]);
309    }
310    Ok((bytes, format!("{:x}", hasher.finalize())))
311}
312
313/// The whole content of an artifact, at most [`INLINE_CAP`] bytes. A
314/// `records` artifact is read up to its last complete line, so a record
315/// being appended is never cut.
316pub fn content(path: &Path, format: &str) -> Result<Value> {
317    let size = fs::metadata(path)?.len();
318    if size > INLINE_CAP as u64 {
319        return Err(coded(
320            "artifact_too_large",
321            format!(
322                "the artifact has {size} bytes, over the {INLINE_CAP}-byte inline cap; {}",
323                if format == "records" {
324                    "read it in pages with offset and limit"
325                } else {
326                    "read the file in the job directory"
327                }
328            ),
329        ));
330    }
331    let mut data = fs::read(path)?;
332    if format == "records" {
333        data.truncate(data.iter().rposition(|b| *b == b'\n').map_or(0, |i| i + 1));
334    }
335    let sha256 = storage::digest(&data);
336    let bytes = data.len();
337    let text = String::from_utf8(data).map_err(|_| "artifact is not UTF-8 text")?;
338    Ok(json!({"format":format,"bytes":bytes,"sha256":sha256,"content":text}))
339}
340
341/// Records `offset..offset + limit` of a `records` artifact: the payloads of
342/// complete records whose checksums hold, at most [`INLINE_CAP`] bytes of
343/// payload (always at least one record). `next_offset` is where the next
344/// page starts; `available` counts the complete records now in the file,
345/// which a running job may extend.
346///
347/// A record is a newline-terminated, non-blank line; a final line without
348/// its newline is still being written and does not count. The file is read
349/// once as a stream: records before `offset` are only counted, and only the
350/// page's records are parsed and checked, so a page costs one pass over the
351/// bytes and memory for the page alone, however long the artifact is.
352pub fn page(path: &Path, offset: usize, limit: usize) -> Result<Value> {
353    if !(1..=MAX_PAGE).contains(&limit) {
354        return Err(coded(
355            "invalid_command",
356            format!("limit must lie in 1..={MAX_PAGE}"),
357        ));
358    }
359    let mut reader = io::BufReader::with_capacity(1 << 16, fs::File::open(path)?);
360    let mut line = Vec::new();
361    let (mut index, mut size, mut full) = (0usize, 0usize, false);
362    let mut records = Vec::new();
363    loop {
364        line.clear();
365        if reader.read_until(b'\n', &mut line)? == 0 || line.last() != Some(&b'\n') {
366            break;
367        }
368        if line.iter().all(u8::is_ascii_whitespace) {
369            continue;
370        }
371        if index >= offset && records.len() < limit && !full {
372            let envelope: storage::Envelope = serde_json::from_slice(&line)
373                .map_err(|e| format!("record {index} of the artifact is no checked record: {e}"))?;
374            if envelope.checksum != storage::digest(envelope.payload_json.as_bytes()) {
375                return Err(format!("record {index} of the artifact fails its checksum").into());
376            }
377            if size + envelope.payload_json.len() > INLINE_CAP {
378                if records.is_empty() {
379                    return Err(coded(
380                        "artifact_too_large",
381                        format!(
382                            "record {index} alone exceeds the {INLINE_CAP}-byte inline cap; read the file in the job directory"
383                        ),
384                    ));
385                }
386                full = true;
387            } else {
388                size += envelope.payload_json.len();
389                records.push(serde_json::from_str::<Value>(&envelope.payload_json)?);
390            }
391        }
392        index += 1;
393    }
394    let next = offset.min(index) + records.len();
395    Ok(
396        json!({"format":"records","offset":offset,"records":records,"next_offset":next,"available":index}),
397    )
398}
optimization_job_api/artifact.rs

optimization_job_api/
artifact.rs
