alphawinnow/
artifact.rs

1use std::{
2    fs::{self, File, OpenOptions},
3    io::{BufRead, BufReader, Write},
4    path::{Path, PathBuf},
5    sync::atomic::{AtomicU64, Ordering},
6};
7
8use serde::{Deserialize, Serialize};
9use thiserror::Error;
10
11use crate::{
12    RejectionReason, SearchSpec, StructuralDescriptor, StructuralScoreWeights,
13    canonical::digest,
14    transform::{Provenance, TransformKind},
15};
16
17pub const CANDIDATE_SCHEMA: u32 = 4;
18pub const MANIFEST_SCHEMA: u32 = 8;
19static TEMPORARY_SEQUENCE: AtomicU64 = AtomicU64::new(0);
20
21/// Portable candidate record. The score contains no market-performance evidence.
22#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
23pub struct CandidateRecord {
24    pub schema: u32,
25    #[serde(default)]
26    pub run_id: String,
27    pub expression: String,
28    pub fingerprint: String,
29    pub semantic_fingerprint: String,
30    pub structural_score: f64,
31    #[serde(default)]
32    pub score_schema: u32,
33    #[serde(default)]
34    pub score_components: StructuralScoreComponents,
35    #[serde(default)]
36    pub normalized_weights: StructuralScoreWeights,
37    #[serde(default)]
38    pub structural_descriptor: StructuralDescriptor,
39    pub root_family: String,
40    pub nodes: usize,
41    pub depth: usize,
42    pub provenance: Provenance,
43}
44
45#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
46pub struct StructuralScoreComponents {
47    pub simplicity: f64,
48    pub archive_novelty: f64,
49    pub family_diversity: f64,
50    pub lineage_diversity: f64,
51    pub structural_novelty: f64,
52    pub transform_diversity: f64,
53}
54
55/// Identity of immutable measured feedback used to guide a search.
56#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
57pub struct FeedbackProvenance {
58    pub dataset_id: String,
59    pub context_checksum: String,
60    pub feedback_checksum: String,
61    pub configuration_checksum: String,
62    pub outcome_label: String,
63}
64
65/// Identity and exact weights of an opt-in feedback-derived operator policy.
66#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
67pub struct OperatorPolicyProvenance {
68    pub configuration_checksum: String,
69    pub base_catalog_checksum: String,
70    pub adjusted_catalog_checksum: String,
71    pub generation_weights: std::collections::BTreeMap<String, u16>,
72    pub transform_action_weights: std::collections::BTreeMap<String, u16>,
73    pub transform_action_source: String,
74    #[serde(default, skip_serializing_if = "Option::is_none")]
75    pub transform_action_archive_checksum: Option<String>,
76}
77
78#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
79pub struct RunManifest {
80    pub schema: u32,
81    pub run_id: String,
82    pub candidate_schema: u32,
83    pub operator_catalog_checksum: String,
84    pub tool_version: String,
85    pub search_spec: SearchSpec,
86    pub config_checksum: String,
87    #[serde(default, skip_serializing_if = "Option::is_none")]
88    pub measured_feedback: Option<FeedbackProvenance>,
89    #[serde(default, skip_serializing_if = "Option::is_none")]
90    pub operator_policy: Option<OperatorPolicyProvenance>,
91    #[serde(default)]
92    pub guided_population_candidates: u64,
93    #[serde(default)]
94    pub seeded_from_archive: bool,
95    #[serde(default)]
96    pub frozen_seed_parents: bool,
97    #[serde(default, skip_serializing_if = "usize_is_zero")]
98    pub feedback_elite_parent_count: usize,
99    #[serde(default)]
100    pub catalog_scaffold_policy: crate::CatalogScaffoldPolicy,
101    pub archive_records: u64,
102    pub archive_accepted_records: u64,
103    pub archive_duplicate_records: u64,
104    pub archive_rejected_records: u64,
105    pub archive_rejection_reasons: std::collections::BTreeMap<RejectionReason, u64>,
106    pub attempted_candidates: u64,
107    pub rejected_invalid_candidates: u64,
108    pub valid_candidates: u64,
109    pub rejected_trivial_candidates: u64,
110    pub rejection_reasons: std::collections::BTreeMap<RejectionReason, u64>,
111    pub duplicate_candidates: u64,
112    pub retained_candidates: u64,
113    pub accepted_transform_counts: std::collections::BTreeMap<TransformKind, u64>,
114    pub retained_transform_counts: std::collections::BTreeMap<TransformKind, u64>,
115    pub generations_completed: u64,
116    pub population_capacity: usize,
117    pub peak_population: usize,
118    pub elite_archive_capacity: usize,
119    pub peak_elite_archive: usize,
120    pub semantic_archive_capacity: usize,
121    pub peak_semantic_archive: usize,
122    pub descriptor_archive_capacity: usize,
123    pub peak_descriptor_archive: usize,
124    pub elapsed_milliseconds: u128,
125    pub termination_reason: String,
126    pub candidate_content_checksum: String,
127}
128
129#[allow(clippy::trivially_copy_pass_by_ref)]
130fn usize_is_zero(value: &usize) -> bool {
131    *value == 0
132}
133
134pub const CHECKPOINT_SCHEMA: u32 = 2;
135
136#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
137pub struct CheckpointDraft {
138    pub expression: String,
139    pub provenance: Provenance,
140    pub semantic_fingerprint: String,
141    pub descriptor: StructuralDescriptor,
142}
143
144#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
145pub struct RunCheckpoint {
146    pub schema: u32,
147    pub run_id: String,
148    pub config_checksum: String,
149    pub operator_catalog_checksum: String,
150    #[serde(default, skip_serializing_if = "Option::is_none")]
151    pub measured_feedback: Option<FeedbackProvenance>,
152    pub next_generation: u64,
153    pub generations_completed: u64,
154    pub attempted_candidates: u64,
155    pub rejected_invalid_candidates: u64,
156    pub duplicate_candidates: u64,
157    pub valid_candidates: u64,
158    pub rejection_reasons: std::collections::BTreeMap<RejectionReason, u64>,
159    pub accepted_transform_counts: std::collections::BTreeMap<TransformKind, u64>,
160    pub semantic_archive: Vec<String>,
161    pub elite_archive: Vec<CheckpointDraft>,
162    pub population: Vec<CheckpointDraft>,
163    pub parent_pool: Vec<CheckpointDraft>,
164    pub peak_population: usize,
165    pub peak_elite_archive: usize,
166    pub peak_semantic_archive: usize,
167    pub termination_state: String,
168}
169
170#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
171pub struct PublicationPointer {
172    pub schema: u32,
173    pub run_id: String,
174    pub candidate_path: PathBuf,
175    pub manifest_path: PathBuf,
176    pub candidate_checksum: String,
177    pub manifest_checksum: String,
178}
179
180#[derive(Debug, Error)]
181pub enum ArtifactError {
182    #[error("cannot read artifact {path}: {source}")]
183    Read {
184        path: PathBuf,
185        source: std::io::Error,
186    },
187    #[error("invalid JSON on line {line} of {path}: {source}")]
188    JsonLine {
189        path: PathBuf,
190        line: usize,
191        source: serde_json::Error,
192    },
193    #[error("unsupported candidate schema {schema} on line {line} of {path}")]
194    Schema {
195        path: PathBuf,
196        line: usize,
197        schema: u32,
198    },
199    #[error("cannot serialize artifact: {0}")]
200    Serialize(#[from] serde_json::Error),
201    #[error("cannot atomically write artifact {path}: {source}")]
202    Write {
203        path: PathBuf,
204        source: std::io::Error,
205    },
206}
207
208/// Read a versioned candidate stream.
209///
210/// # Errors
211/// Returns an error for I/O failures, malformed JSON, or unsupported schemas.
212pub fn read_candidates(path: &Path) -> Result<Vec<CandidateRecord>, ArtifactError> {
213    let file = File::open(path).map_err(|source| ArtifactError::Read {
214        path: path.to_owned(),
215        source,
216    })?;
217    let mut records = Vec::new();
218    for (index, line) in BufReader::new(file).lines().enumerate() {
219        let line_number = index + 1;
220        let line = line.map_err(|source| ArtifactError::Read {
221            path: path.to_owned(),
222            source,
223        })?;
224        if line.trim().is_empty() {
225            continue;
226        }
227        let record: CandidateRecord =
228            serde_json::from_str(&line).map_err(|source| ArtifactError::JsonLine {
229                path: path.to_owned(),
230                line: line_number,
231                source,
232            })?;
233        if record.schema == 0 || record.schema > CANDIDATE_SCHEMA {
234            return Err(ArtifactError::Schema {
235                path: path.to_owned(),
236                line: line_number,
237                schema: record.schema,
238            });
239        }
240        records.push(record);
241    }
242    Ok(records)
243}
244
245/// Serialize records exactly as they will appear in JSONL.
246///
247/// # Errors
248/// Returns an error if a record cannot be serialized.
249pub fn candidate_content(records: &[CandidateRecord]) -> Result<String, ArtifactError> {
250    let mut content = String::new();
251    for record in records {
252        content.push_str(&serde_json::to_string(record)?);
253        content.push('\n');
254    }
255    Ok(content)
256}
257
258/// Serialize and atomically replace a candidate artifact, returning its checksum.
259///
260/// # Errors
261/// Returns an error on serialization, temporary-file, sync, or rename failures.
262pub fn write_jsonl_atomic(
263    path: &Path,
264    records: &[CandidateRecord],
265) -> Result<String, ArtifactError> {
266    let content = candidate_content(records)?;
267    atomic_write(path, content.as_bytes())?;
268    Ok(digest(&content))
269}
270
271/// Atomically serialize a run manifest.
272///
273/// # Errors
274/// Returns an error on serialization, temporary-file, sync, or rename failures.
275pub fn write_manifest_atomic(path: &Path, manifest: &RunManifest) -> Result<(), ArtifactError> {
276    let mut content = serde_json::to_vec_pretty(manifest)?;
277    content.push(b'\n');
278    atomic_write(path, &content)
279}
280
281/// Atomically write a deterministic resume checkpoint.
282///
283/// # Errors
284/// Returns an error on serialization or durable replacement failure.
285pub fn write_checkpoint_atomic(
286    path: &Path,
287    checkpoint: &RunCheckpoint,
288) -> Result<(), ArtifactError> {
289    let mut content = serde_json::to_vec(checkpoint)?;
290    content.push(b'\n');
291    atomic_write(path, &content)
292}
293
294/// Read a versioned resume checkpoint.
295///
296/// # Errors
297/// Returns an error for I/O, JSON, or unsupported checkpoint schema failures.
298pub fn read_checkpoint(path: &Path) -> Result<RunCheckpoint, ArtifactError> {
299    let content = fs::read(path).map_err(|source| ArtifactError::Read {
300        path: path.to_owned(),
301        source,
302    })?;
303    let checkpoint: RunCheckpoint =
304        serde_json::from_slice(&content).map_err(|source| ArtifactError::JsonLine {
305            path: path.to_owned(),
306            line: 1,
307            source,
308        })?;
309    if checkpoint.schema != CHECKPOINT_SCHEMA {
310        return Err(ArtifactError::Schema {
311            path: path.to_owned(),
312            line: 1,
313            schema: checkpoint.schema,
314        });
315    }
316    Ok(checkpoint)
317}
318
319#[must_use]
320pub fn manifest_path(output: &Path) -> PathBuf {
321    let name = output
322        .file_name()
323        .map_or_else(|| "candidates".into(), std::ffi::OsStr::to_os_string);
324    let mut adjacent = name;
325    adjacent.push(".manifest.json");
326    output.with_file_name(adjacent)
327}
328
329#[must_use]
330pub fn publication_pointer_path(output: &Path) -> PathBuf {
331    let name = output
332        .file_name()
333        .map_or_else(|| "candidates".into(), std::ffi::OsStr::to_os_string);
334    let mut adjacent = name;
335    adjacent.push(".current.json");
336    output.with_file_name(adjacent)
337}
338
339/// Flush a directory entry so a completed rename survives a crash.
340///
341/// Opening a directory as a file to flush it is a Unix idiom. Windows rejects
342/// that open with `ERROR_ACCESS_DENIED` unless the handle carries backup
343/// semantics, and NTFS journals the metadata operation regardless, so there is
344/// nothing to flush by hand there. Keeping the call strict on Unix preserves
345/// the durability the publication protocol documents without failing a
346/// publication that actually succeeded on other platforms.
347// Off Unix the body has nothing that can fail, so the `Result` looks
348// redundant there; it is the signature that keeps the Unix caller strict.
349#[cfg_attr(not(unix), allow(clippy::unnecessary_wraps))]
350fn sync_directory(path: &Path) -> std::io::Result<()> {
351    #[cfg(unix)]
352    {
353        File::open(path)?.sync_all()?;
354    }
355    #[cfg(not(unix))]
356    {
357        let _ = path;
358    }
359    Ok(())
360}
361
362/// Durably publish an immutable candidate/manifest pair, then atomically move
363/// one pointer to the complete version. Compatibility aliases are refreshed
364/// only after the authoritative pair is durable.
365///
366/// # Errors
367/// Returns an error for inconsistent run IDs, serialization, or filesystem
368/// durability failures.
369pub fn write_run_transactional(
370    output: &Path,
371    records: &[CandidateRecord],
372    manifest: &RunManifest,
373) -> Result<PublicationPointer, ArtifactError> {
374    if records
375        .iter()
376        .any(|record| record.run_id != manifest.run_id)
377    {
378        return Err(ArtifactError::Write {
379            path: output.to_owned(),
380            source: std::io::Error::new(
381                std::io::ErrorKind::InvalidData,
382                "candidate and manifest run IDs differ",
383            ),
384        });
385    }
386    let candidate_bytes = candidate_content(records)?.into_bytes();
387    let candidate_checksum = digest(std::str::from_utf8(&candidate_bytes).unwrap_or_default());
388    if candidate_checksum != manifest.candidate_content_checksum {
389        return Err(ArtifactError::Write {
390            path: output.to_owned(),
391            source: std::io::Error::new(
392                std::io::ErrorKind::InvalidData,
393                "candidate checksum does not match manifest",
394            ),
395        });
396    }
397    let mut manifest_bytes = serde_json::to_vec_pretty(manifest)?;
398    manifest_bytes.push(b'\n');
399    let manifest_checksum = digest(std::str::from_utf8(&manifest_bytes).unwrap_or_default());
400    let parent = output.parent().unwrap_or_else(|| Path::new("."));
401    let file_name = output.file_name().unwrap_or_default().to_string_lossy();
402    let runs = parent.join(format!(".{file_name}.runs"));
403    fs::create_dir_all(&runs).map_err(|source| ArtifactError::Write {
404        path: runs.clone(),
405        source,
406    })?;
407    let version = format!(
408        "{}-{}-{}",
409        &manifest.run_id[..manifest.run_id.len().min(16)],
410        &candidate_checksum[..candidate_checksum.len().min(16)],
411        &manifest_checksum[..manifest_checksum.len().min(16)]
412    );
413    let final_directory = runs.join(version);
414    if !final_directory.exists() {
415        let sequence = TEMPORARY_SEQUENCE.fetch_add(1, Ordering::Relaxed);
416        let staging = runs.join(format!(".stage-{}-{sequence}", std::process::id()));
417        fs::create_dir(&staging).map_err(|source| ArtifactError::Write {
418            path: staging.clone(),
419            source,
420        })?;
421        let result = (|| -> Result<(), std::io::Error> {
422            write_new_file(&staging.join("candidates.jsonl"), &candidate_bytes)?;
423            write_new_file(&staging.join("manifest.json"), &manifest_bytes)?;
424            sync_directory(&staging)?;
425            fs::rename(&staging, &final_directory)?;
426            sync_directory(&runs)?;
427            Ok(())
428        })();
429        if let Err(source) = result {
430            let _ = fs::remove_dir_all(&staging);
431            return Err(ArtifactError::Write {
432                path: final_directory,
433                source,
434            });
435        }
436    }
437    let pointer = PublicationPointer {
438        schema: 1,
439        run_id: manifest.run_id.clone(),
440        candidate_path: final_directory.join("candidates.jsonl"),
441        manifest_path: final_directory.join("manifest.json"),
442        candidate_checksum,
443        manifest_checksum,
444    };
445    let mut pointer_bytes = serde_json::to_vec_pretty(&pointer)?;
446    pointer_bytes.push(b'\n');
447    atomic_write(&publication_pointer_path(output), &pointer_bytes)?;
448    write_jsonl_atomic(output, records)?;
449    write_manifest_atomic(&manifest_path(output), manifest)?;
450    Ok(pointer)
451}
452
453/// Read the immutable pair selected by the authoritative publication pointer.
454///
455/// # Errors
456/// Returns an error for a missing/inconsistent pointer, candidate stream, or
457/// manifest.
458pub fn read_published_run(
459    output: &Path,
460) -> Result<(Vec<CandidateRecord>, RunManifest), ArtifactError> {
461    let pointer_path = publication_pointer_path(output);
462    let pointer_bytes = fs::read(&pointer_path).map_err(|source| ArtifactError::Read {
463        path: pointer_path.clone(),
464        source,
465    })?;
466    let pointer: PublicationPointer =
467        serde_json::from_slice(&pointer_bytes).map_err(|source| ArtifactError::JsonLine {
468            path: pointer_path,
469            line: 1,
470            source,
471        })?;
472    if pointer.schema != 1 {
473        return Err(ArtifactError::Schema {
474            path: publication_pointer_path(output),
475            line: 1,
476            schema: pointer.schema,
477        });
478    }
479    let candidate_bytes =
480        fs::read(&pointer.candidate_path).map_err(|source| ArtifactError::Read {
481            path: pointer.candidate_path.clone(),
482            source,
483        })?;
484    let records = read_candidates(&pointer.candidate_path)?;
485    let manifest_bytes =
486        fs::read(&pointer.manifest_path).map_err(|source| ArtifactError::Read {
487            path: pointer.manifest_path.clone(),
488            source,
489        })?;
490    let manifest: RunManifest =
491        serde_json::from_slice(&manifest_bytes).map_err(|source| ArtifactError::JsonLine {
492            path: pointer.manifest_path,
493            line: 1,
494            source,
495        })?;
496    if manifest.run_id != pointer.run_id
497        || records.iter().any(|record| record.run_id != pointer.run_id)
498    {
499        return Err(ArtifactError::Write {
500            path: output.to_owned(),
501            source: std::io::Error::new(
502                std::io::ErrorKind::InvalidData,
503                "published pair has inconsistent run IDs",
504            ),
505        });
506    }
507    let candidate_checksum = digest(std::str::from_utf8(&candidate_bytes).unwrap_or_default());
508    let manifest_checksum = digest(std::str::from_utf8(&manifest_bytes).unwrap_or_default());
509    if candidate_checksum != pointer.candidate_checksum
510        || candidate_checksum != manifest.candidate_content_checksum
511    {
512        return Err(ArtifactError::Write {
513            path: output.to_owned(),
514            source: std::io::Error::new(
515                std::io::ErrorKind::InvalidData,
516                "published candidate checksum verification failed",
517            ),
518        });
519    }
520    if manifest_checksum != pointer.manifest_checksum {
521        return Err(ArtifactError::Write {
522            path: output.to_owned(),
523            source: std::io::Error::new(
524                std::io::ErrorKind::InvalidData,
525                "published manifest checksum verification failed",
526            ),
527        });
528    }
529    Ok((records, manifest))
530}
531
532fn write_new_file(path: &Path, content: &[u8]) -> Result<(), std::io::Error> {
533    let mut file = OpenOptions::new().write(true).create_new(true).open(path)?;
534    file.write_all(content)?;
535    file.sync_all()
536}
537
538pub(crate) fn atomic_write(path: &Path, content: &[u8]) -> Result<(), ArtifactError> {
539    let parent = path.parent().unwrap_or_else(|| Path::new("."));
540    let file_name = path.file_name().unwrap_or_default().to_string_lossy();
541    let sequence = TEMPORARY_SEQUENCE.fetch_add(1, Ordering::Relaxed);
542    let temporary = parent.join(format!(
543        ".{file_name}.{}.{sequence}.tmp",
544        std::process::id()
545    ));
546    let result = (|| -> Result<(), std::io::Error> {
547        let mut file = OpenOptions::new()
548            .write(true)
549            .create_new(true)
550            .open(&temporary)?;
551        file.write_all(content)?;
552        file.sync_all()?;
553        drop(file);
554        fs::rename(&temporary, path)?;
555        let _ = sync_directory(parent);
556        Ok(())
557    })();
558    if let Err(source) = result {
559        let _ = fs::remove_file(&temporary);
560        return Err(ArtifactError::Write {
561            path: path.to_owned(),
562            source,
563        });
564    }
565    Ok(())
566}
567
568#[cfg(test)]
569mod tests {
570    use tempfile::tempdir;
571
572    use super::*;
573    use crate::transform::{Provenance, TransformKind};
574
575    fn record(expression: &str) -> CandidateRecord {
576        CandidateRecord {
577            schema: CANDIDATE_SCHEMA,
578            run_id: "test-run".to_owned(),
579            expression: expression.to_owned(),
580            fingerprint: "a".repeat(64),
581            semantic_fingerprint: "b".repeat(64),
582            structural_score: 0.5,
583            score_schema: 2,
584            score_components: StructuralScoreComponents::default(),
585            normalized_weights: StructuralScoreWeights::default(),
586            structural_descriptor: StructuralDescriptor::default(),
587            root_family: "field".to_owned(),
588            nodes: 1,
589            depth: 1,
590            provenance: Provenance {
591                kind: TransformKind::Initial,
592                operation: "test".to_owned(),
593                parent_fingerprints: Vec::new(),
594                requested_operation: None,
595                affected_path: None,
596                old_subtree_fingerprint: None,
597                new_subtree_fingerprint: None,
598                retry_count: 0,
599            },
600        }
601    }
602
603    #[test]
604    fn atomic_jsonl_replaces_complete_artifact_and_leaves_no_temp() {
605        let directory = tempdir().unwrap();
606        let path = directory.path().join("out.jsonl");
607        fs::write(&path, "old\n").unwrap();
608        write_jsonl_atomic(&path, &[record("close")]).unwrap();
609        assert_eq!(read_candidates(&path).unwrap()[0].expression, "close");
610        assert_eq!(fs::read_dir(directory.path()).unwrap().count(), 1);
611    }
612}
alphawinnow/artifact.rs

alphawinnow/
artifact.rs
