diff --git a/crates/renderflow-core/src/artifact/cache.rs b/crates/renderflow-core/src/artifact/cache.rs new file mode 100644 index 0000000..386e1c3 --- /dev/null +++ b/crates/renderflow-core/src/artifact/cache.rs @@ -0,0 +1,141 @@ +use std::collections::HashMap; +use std::fs; +use std::io::Write; +use std::path::Path; + +use anyhow::{Context, Result}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use tracing::warn; + +use super::Artifact; +use crate::graph::Format; + +/// Artifact-native DAG cache mapping deterministic node keys to stored artifacts. +#[derive(Debug, Default, Serialize, Deserialize)] +pub struct ArtifactCache { + entries: HashMap, +} + +impl ArtifactCache { + pub(crate) fn get(&self, key: &str) -> Option<&Artifact> { + self.entries.get(key) + } + + pub(crate) fn insert(&mut self, key: String, artifact: Artifact) { + self.entries.insert(key, artifact); + } +} + +/// Compute a DAG cache key from artifact identity and transform configuration. +/// +/// Large payload bytes are deliberately not re-hashed here: the input artifact's +/// SHA-256 digest and byte size already establish content identity. +pub fn compute_artifact_node_hash( + input: &Artifact, + from: Format, + to: Format, + transform_identity: &str, +) -> String { + let mut hasher = Sha256::new(); + hasher.update(input.id().as_str().as_bytes()); + hasher.update(b"\x00artifact-id\x00"); + hasher.update(input.digest().algorithm().to_string().as_bytes()); + hasher.update(b"\x00digest\x00"); + hasher.update(input.digest().value().as_bytes()); + hasher.update(b"\x00size\x00"); + hasher.update(input.size_bytes().to_le_bytes()); + hasher.update(b"\x00from\x00"); + hasher.update(from.to_string().as_bytes()); + hasher.update(b"\x00to\x00"); + hasher.update(to.to_string().as_bytes()); + hasher.update(b"\x00transform\x00"); + hasher.update(transform_identity.as_bytes()); + format!("{:x}", hasher.finalize()) +} + +pub(crate) fn load_artifact_cache(path: &Path) -> ArtifactCache { + if !path.exists() { + return ArtifactCache::default(); + } + match fs::read_to_string(path) { + Ok(content) => match serde_json::from_str(&content) { + Ok(cache) => cache, + Err(error) => { + warn!( + path = %path.display(), + error = %error, + "Artifact DAG cache is unreadable or uses a legacy schema; starting empty" + ); + ArtifactCache::default() + } + }, + Err(error) => { + warn!( + path = %path.display(), + error = %error, + "Failed to read artifact DAG cache; starting empty" + ); + ArtifactCache::default() + } + } +} + +pub(crate) fn save_artifact_cache(cache: &ArtifactCache, path: &Path) -> Result<()> { + let parent = path + .parent() + .filter(|candidate| !candidate.as_os_str().is_empty()) + .unwrap_or_else(|| Path::new(".")); + fs::create_dir_all(parent).with_context(|| { + format!( + "Failed to create artifact cache directory '{}'", + parent.display() + ) + })?; + + let mut temporary = tempfile::NamedTempFile::new_in(parent) + .context("Failed to create artifact cache temporary file")?; + serde_json::to_writer(&mut temporary, cache).context("Failed to serialize artifact cache")?; + temporary.flush().context("Failed to flush artifact cache")?; + temporary + .as_file() + .sync_all() + .context("Failed to sync artifact cache")?; + temporary + .persist(path) + .map_err(|error| error.error) + .with_context(|| format!("Failed to atomically save artifact cache '{}'", path.display()))?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::artifact::{ArtifactDescriptor, ArtifactStorageClass, ArtifactStore}; + + #[test] + fn cache_key_uses_artifact_identity_not_payload_buffer() { + let directory = tempfile::tempdir().unwrap(); + let store = ArtifactStore::new(directory.path()).unwrap(); + let artifact = store + .put_bytes( + &[0, 255, 1, 2], + ArtifactDescriptor::for_format(Format::Png, ArtifactStorageClass::Source), + ) + .unwrap(); + let first = compute_artifact_node_hash( + &artifact, + Format::Png, + Format::Webp, + "adapter:v1:quality=90", + ); + let second = compute_artifact_node_hash( + &artifact, + Format::Png, + Format::Webp, + "adapter:v1:quality=80", + ); + assert_ne!(first, second); + assert_eq!(first.len(), 64); + } +} diff --git a/crates/renderflow-core/src/artifact/mod.rs b/crates/renderflow-core/src/artifact/mod.rs new file mode 100644 index 0000000..073e70a --- /dev/null +++ b/crates/renderflow-core/src/artifact/mod.rs @@ -0,0 +1,20 @@ +//! Binary-safe artifact kernel. +//! +//! The artifact kernel separates payload storage from transform orchestration. +//! Payloads are content-addressed, file-backed, and identified by SHA-256 so +//! documents, images, audio, video, archives, and structured data can traverse +//! the same graph without a UTF-8 assumption. + +mod cache; +mod model; +mod store; +mod transform; + +pub use cache::compute_artifact_node_hash; +pub(crate) use cache::{load_artifact_cache, save_artifact_cache, ArtifactCache}; +pub use model::{ + Artifact, ArtifactCollection, ArtifactDescriptor, ArtifactDigest, ArtifactId, ArtifactPayload, + ArtifactStorageClass, CanonicalFormat, DigestAlgorithm, MediaType, +}; +pub use store::ArtifactStore; +pub use transform::{ArtifactTransform, TextTransformAdapter}; diff --git a/crates/renderflow-core/src/artifact/model.rs b/crates/renderflow-core/src/artifact/model.rs new file mode 100644 index 0000000..307f2f4 --- /dev/null +++ b/crates/renderflow-core/src/artifact/model.rs @@ -0,0 +1,501 @@ +use std::collections::BTreeMap; +use std::fmt; +use std::path::{Path, PathBuf}; + +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use sha2::{Digest, Sha256}; + +use crate::graph::Format; + +/// Stable identifier for an artifact record. +/// +/// Payload bytes are content-addressed independently by [`ArtifactDigest`]. The +/// record identifier additionally commits to canonical format and ordered source +/// lineage, so byte-preserving conversions remain distinct provenance records. +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(transparent)] +pub struct ArtifactId(String); + +impl ArtifactId { + pub(crate) fn from_record( + digest: &ArtifactDigest, + format: &CanonicalFormat, + sources: &[ArtifactId], + ) -> Self { + let mut hasher = Sha256::new(); + hasher.update(digest.algorithm().to_string().as_bytes()); + hasher.update(b"\x00digest\x00"); + hasher.update(digest.value().as_bytes()); + hasher.update(b"\x00format\x00"); + hasher.update(format.as_str().as_bytes()); + hasher.update(b"\x00sources\x00"); + for source in sources { + hasher.update(source.as_str().as_bytes()); + hasher.update(b"\x00"); + } + Self(format!("artifact:sha256:{:x}", hasher.finalize())) + } + + /// Return the stable identifier as a string. + pub fn as_str(&self) -> &str { + &self.0 + } +} + +impl fmt::Display for ArtifactId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&self.0) + } +} + +/// Digest algorithm used for content identity. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum DigestAlgorithm { + /// SHA-256, the canonical Renderflow artifact digest algorithm. + Sha256, +} + +impl fmt::Display for DigestAlgorithm { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Sha256 => f.write_str("sha256"), + } + } +} + +/// Content digest for an artifact payload. +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct ArtifactDigest { + algorithm: DigestAlgorithm, + value: String, +} + +impl ArtifactDigest { + pub(crate) fn sha256(value: String) -> Self { + Self { + algorithm: DigestAlgorithm::Sha256, + value, + } + } + + /// Digest algorithm. + pub fn algorithm(&self) -> DigestAlgorithm { + self.algorithm + } + + /// Lowercase hexadecimal digest value. + pub fn value(&self) -> &str { + &self.value + } +} + +impl fmt::Display for ArtifactDigest { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}:{}", self.algorithm, self.value) + } +} + +/// Stable, tool-neutral format identifier carried by an artifact. +/// +/// The artifact kernel intentionally stores the canonical string form rather +/// than duplicating the graph's format enum in serialized cache/evidence data. +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(transparent)] +pub struct CanonicalFormat(String); + +impl CanonicalFormat { + /// Construct a canonical format from a non-empty identifier. + pub fn new(value: impl Into) -> anyhow::Result { + let value = value.into(); + if value.trim().is_empty() { + anyhow::bail!("artifact format must not be empty"); + } + Ok(Self(value)) + } + + /// Return the canonical identifier. + pub fn as_str(&self) -> &str { + &self.0 + } +} + +impl From for CanonicalFormat { + fn from(value: Format) -> Self { + Self(value.to_string()) + } +} + +impl fmt::Display for CanonicalFormat { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&self.0) + } +} + +/// MIME/media type carried with an artifact. +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(transparent)] +pub struct MediaType(String); + +impl MediaType { + /// Construct a media type from a non-empty value. + pub fn new(value: impl Into) -> anyhow::Result { + let value = value.into(); + if value.trim().is_empty() { + anyhow::bail!("artifact media type must not be empty"); + } + Ok(Self(value)) + } + + /// Return the media type string. + pub fn as_str(&self) -> &str { + &self.0 + } + + /// Return Renderflow's default media type for a graph [`Format`]. + pub fn for_format(format: Format) -> Self { + let value = match format { + Format::Markdown => "text/markdown", + Format::Html => "text/html", + Format::Pdf => "application/pdf", + Format::Docx => { + "application/vnd.openxmlformats-officedocument.wordprocessingml.document" + } + Format::Epub => "application/epub+zip", + Format::Rst => "text/x-rst", + Format::Latex => "application/x-latex", + Format::Fountain => "text/plain", + Format::Jpeg => "image/jpeg", + Format::Png => "image/png", + Format::Tiff => "image/tiff", + Format::Webp => "image/webp", + Format::Gif => "image/gif", + Format::Bmp => "image/bmp", + Format::Avif => "image/avif", + Format::Svg => "image/svg+xml", + Format::Cbz => "application/vnd.comicbook+zip", + Format::Wav | Format::Bwf => "audio/wav", + Format::Aiff => "audio/aiff", + Format::Pcm => "application/octet-stream", + Format::Flac => "audio/flac", + Format::M4aAlac | Format::M4aAac => "audio/mp4", + Format::Wv => "audio/x-wavpack", + Format::Ape => "audio/ape", + Format::Tta => "audio/x-tta", + Format::Dsf | Format::Dff => "audio/dsd", + Format::Shn => "audio/x-shorten", + Format::Mp3 | Format::Mp2 => "audio/mpeg", + Format::Aac => "audio/aac", + Format::Ogg => "audio/ogg", + Format::Opus => "audio/opus", + Format::Wma => "audio/x-ms-wma", + Format::Amr => "audio/amr", + Format::Ra => "audio/vnd.rn-realaudio", + Format::Oma => "audio/atrac", + Format::Ac3 => "audio/ac3", + Format::Ec3 => "audio/eac3", + Format::Thd => "audio/vnd.dolby.mlp", + Format::Dts => "audio/vnd.dts", + Format::DtsHd => "audio/vnd.dts.hd", + Format::Midi => "audio/midi", + Format::Mod => "audio/mod", + Format::Mp4 => "video/mp4", + Format::Mov => "video/quicktime", + Format::Mkv => "video/x-matroska", + Format::WebM => "video/webm", + Format::Avi => "video/x-msvideo", + Format::Json => "application/json", + Format::Yaml => "application/yaml", + Format::Toml => "application/toml", + Format::Csv => "text/csv", + Format::Tsv => "text/tab-separated-values", + Format::Xml => "application/xml", + Format::Zip => "application/zip", + Format::TarGz => "application/gzip", + Format::TarXz => "application/x-xz", + Format::Srt => "application/x-subrip", + Format::WebVtt => "text/vtt", + }; + Self(value.to_string()) + } +} + +impl fmt::Display for MediaType { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&self.0) + } +} + +/// Lifecycle/storage role of an artifact inside an execution. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ArtifactStorageClass { + /// Immutable source copied into the content-addressed store before execution. + Source, + /// Durable work product used by downstream graph edges. + Intermediate, + /// Artifact selected for final materialization/publication. + Terminal, + /// Artifact reused from a previous execution cache entry. + Cached, + /// Short-lived artifact that may be garbage-collected after the run. + Ephemeral, +} + +/// File-backed payload handle relative to an [`ArtifactStore`](super::ArtifactStore). +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct ArtifactPayload { + relative_path: PathBuf, +} + +impl ArtifactPayload { + pub(crate) fn new(relative_path: PathBuf) -> Self { + Self { relative_path } + } + + /// Store-relative payload path. Resolve it through [`ArtifactStore`](super::ArtifactStore) + /// rather than joining it manually. + pub fn relative_path(&self) -> &Path { + &self.relative_path + } +} + +/// First-class binary-safe artifact record. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct Artifact { + id: ArtifactId, + format: CanonicalFormat, + media_type: MediaType, + digest: ArtifactDigest, + size_bytes: u64, + payload: ArtifactPayload, + #[serde(default)] + metadata: BTreeMap, + #[serde(default)] + sources: Vec, + storage_class: ArtifactStorageClass, +} + +impl Artifact { + #[allow(clippy::too_many_arguments)] + pub(crate) fn new( + id: ArtifactId, + format: CanonicalFormat, + media_type: MediaType, + digest: ArtifactDigest, + size_bytes: u64, + payload: ArtifactPayload, + metadata: BTreeMap, + sources: Vec, + storage_class: ArtifactStorageClass, + ) -> Self { + Self { + id, + format, + media_type, + digest, + size_bytes, + payload, + metadata, + sources, + storage_class, + } + } + + /// Stable artifact identifier. + pub fn id(&self) -> &ArtifactId { + &self.id + } + + /// Canonical Renderflow format identifier. + pub fn format(&self) -> &CanonicalFormat { + &self.format + } + + /// MIME/media type. + pub fn media_type(&self) -> &MediaType { + &self.media_type + } + + /// SHA-256 content digest. + pub fn digest(&self) -> &ArtifactDigest { + &self.digest + } + + /// Payload size in bytes. + pub fn size_bytes(&self) -> u64 { + self.size_bytes + } + + /// File-backed payload handle. + pub fn payload(&self) -> &ArtifactPayload { + &self.payload + } + + /// Structured artifact metadata. + pub fn metadata(&self) -> &BTreeMap { + &self.metadata + } + + /// Ordered parent/source artifact identifiers. + pub fn sources(&self) -> &[ArtifactId] { + &self.sources + } + + /// Lifecycle/storage role. + pub fn storage_class(&self) -> ArtifactStorageClass { + self.storage_class + } + + pub(crate) fn with_storage_class(mut self, storage_class: ArtifactStorageClass) -> Self { + self.storage_class = storage_class; + self + } +} + +/// Metadata used when importing or writing a payload into the artifact store. +#[derive(Debug, Clone)] +pub struct ArtifactDescriptor { + pub(crate) format: CanonicalFormat, + pub(crate) media_type: MediaType, + pub(crate) storage_class: ArtifactStorageClass, + pub(crate) metadata: BTreeMap, + pub(crate) sources: Vec, +} + +impl ArtifactDescriptor { + /// Create a descriptor using Renderflow's canonical format/media-type mapping. + pub fn for_format(format: Format, storage_class: ArtifactStorageClass) -> Self { + Self { + format: format.into(), + media_type: MediaType::for_format(format), + storage_class, + metadata: BTreeMap::new(), + sources: Vec::new(), + } + } + + /// Override the media type when a provider has more precise information. + pub fn with_media_type(mut self, media_type: MediaType) -> Self { + self.media_type = media_type; + self + } + + /// Add one ordered parent/source relationship. + pub fn with_source(mut self, source: ArtifactId) -> Self { + self.sources.push(source); + self + } + + /// Replace the ordered source relationship list. + pub fn with_sources(mut self, sources: impl IntoIterator) -> Self { + self.sources = sources.into_iter().collect(); + self + } + + /// Attach structured metadata. + pub fn with_metadata(mut self, key: impl Into, value: impl Into) -> Self { + self.metadata.insert(key.into(), value.into()); + self + } +} + +/// Ordered collection of artifacts used by aggregation transforms. +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +pub struct ArtifactCollection { + artifacts: Vec, +} + +impl ArtifactCollection { + /// Create an ordered collection. + pub fn new(artifacts: Vec) -> Self { + Self { artifacts } + } + + /// Create a one-element collection. + pub fn one(artifact: Artifact) -> Self { + Self::new(vec![artifact]) + } + + /// Number of artifacts in the collection. + pub fn len(&self) -> usize { + self.artifacts.len() + } + + /// Whether the collection contains no artifacts. + pub fn is_empty(&self) -> bool { + self.artifacts.is_empty() + } + + /// Iterate in the declared input order. + pub fn iter(&self) -> impl ExactSizeIterator { + self.artifacts.iter() + } + + /// Borrow the ordered artifacts. + pub fn as_slice(&self) -> &[Artifact] { + &self.artifacts + } + + /// Consume the collection into its ordered artifacts. + pub fn into_vec(self) -> Vec { + self.artifacts + } + + /// Return the only artifact, or an error when the collection is not singular. + pub fn into_one(self) -> anyhow::Result { + if self.artifacts.len() != 1 { + anyhow::bail!( + "expected exactly one artifact, found {}", + self.artifacts.len() + ); + } + Ok(self + .artifacts + .into_iter() + .next() + .expect("length checked above")) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn test_artifact(digest_value: char) -> Artifact { + let digest = ArtifactDigest::sha256(digest_value.to_string().repeat(64)); + let format: CanonicalFormat = Format::Png.into(); + let id = ArtifactId::from_record(&digest, &format, &[]); + Artifact::new( + id, + format, + MediaType::for_format(Format::Png), + digest, + 1, + ArtifactPayload::new(format!("objects/{digest_value}").into()), + BTreeMap::new(), + Vec::new(), + ArtifactStorageClass::Source, + ) + } + + #[test] + fn collection_preserves_input_order() { + let artifact_a = test_artifact('a'); + let artifact_b = test_artifact('b'); + let collection = ArtifactCollection::new(vec![artifact_a.clone(), artifact_b.clone()]); + let ids: Vec<_> = collection.iter().map(|artifact| artifact.id()).collect(); + assert_eq!(ids, vec![artifact_a.id(), artifact_b.id()]); + } + + #[test] + fn record_identity_includes_format_and_lineage() { + let digest = ArtifactDigest::sha256("a".repeat(64)); + let png: CanonicalFormat = Format::Png.into(); + let webp: CanonicalFormat = Format::Webp.into(); + let source = ArtifactId::from_record(&digest, &png, &[]); + let derived = ArtifactId::from_record(&digest, &webp, std::slice::from_ref(&source)); + assert_ne!(source, derived); + } +} diff --git a/crates/renderflow-core/src/artifact/store.rs b/crates/renderflow-core/src/artifact/store.rs new file mode 100644 index 0000000..6a390d5 --- /dev/null +++ b/crates/renderflow-core/src/artifact/store.rs @@ -0,0 +1,337 @@ +use std::fs::{self, File}; +use std::io::{self, Cursor, Read, Write}; +use std::path::{Component, Path, PathBuf}; + +use anyhow::{Context, Result}; +use sha2::{Digest, Sha256}; + +use super::{Artifact, ArtifactDescriptor, ArtifactDigest, ArtifactId, ArtifactPayload}; + +const COPY_BUFFER_BYTES: usize = 64 * 1024; + +/// File-backed content-addressed artifact store. +/// +/// Payloads are streamed into a temporary file while a SHA-256 digest is +/// computed, then atomically promoted to a deterministic object path. Canonical +/// source files are therefore never mutated in place. +#[derive(Debug, Clone)] +pub struct ArtifactStore { + root: PathBuf, +} + +impl ArtifactStore { + /// Open or create a content-addressed store at `root`. + pub fn new(root: impl Into) -> Result { + let root = root.into(); + fs::create_dir_all(root.join("objects/sha256")) + .with_context(|| format!("Failed to create artifact store at '{}'", root.display()))?; + fs::create_dir_all(root.join(".tmp")).with_context(|| { + format!( + "Failed to create artifact store temporary directory at '{}'", + root.display() + ) + })?; + Ok(Self { root }) + } + + /// Store root. + pub fn root(&self) -> &Path { + &self.root + } + + /// Import an immutable source/intermediate file without assuming UTF-8. + pub fn import_path( + &self, + path: impl AsRef, + descriptor: ArtifactDescriptor, + ) -> Result { + let path = path.as_ref(); + let file = File::open(path) + .with_context(|| format!("Failed to open artifact source '{}'", path.display()))?; + self.put_reader(file, descriptor) + .with_context(|| format!("Failed to import artifact source '{}'", path.display())) + } + + /// Store an in-memory byte slice. + pub fn put_bytes(&self, bytes: &[u8], descriptor: ArtifactDescriptor) -> Result { + self.put_reader(Cursor::new(bytes), descriptor) + } + + /// Stream arbitrary bytes into the content-addressed store. + /// + /// The entire payload is never required to reside in memory. The temporary + /// object is promoted only after the digest and size are known and all bytes + /// have been written successfully. + pub fn put_reader( + &self, + mut reader: R, + descriptor: ArtifactDescriptor, + ) -> Result { + let mut temporary = tempfile::NamedTempFile::new_in(self.temporary_directory()) + .context("Failed to create artifact-store temporary file")?; + let mut hasher = Sha256::new(); + let mut size_bytes = 0_u64; + let mut buffer = [0_u8; COPY_BUFFER_BYTES]; + + loop { + let read = reader + .read(&mut buffer) + .context("Failed while reading artifact payload")?; + if read == 0 { + break; + } + hasher.update(&buffer[..read]); + temporary + .write_all(&buffer[..read]) + .context("Failed while writing artifact-store temporary file")?; + size_bytes = size_bytes + .checked_add(read as u64) + .context("Artifact size overflowed u64")?; + } + + temporary + .flush() + .context("Failed to flush artifact-store temporary file")?; + temporary + .as_file() + .sync_all() + .context("Failed to sync artifact-store temporary file")?; + + let digest_hex = format!("{:x}", hasher.finalize()); + let digest = ArtifactDigest::sha256(digest_hex.clone()); + let relative_path = Self::object_relative_path(&digest_hex); + let target = self.root.join(&relative_path); + + if let Some(parent) = target.parent() { + fs::create_dir_all(parent).with_context(|| { + format!( + "Failed to create artifact object directory '{}'", + parent.display() + ) + })?; + } + + if !target.exists() { + match temporary.persist(&target) { + Ok(file) => { + file.sync_all().with_context(|| { + format!("Failed to sync artifact object '{}'", target.display()) + })?; + } + Err(error) if target.exists() => { + // A concurrent writer won the race for the same digest. + drop(error.file); + } + Err(error) => { + return Err(error.error).with_context(|| { + format!("Failed to persist artifact object '{}'", target.display()) + }) + } + } + } + + let id = ArtifactId::from_record(&digest, &descriptor.format, &descriptor.sources); + Ok(Artifact::new( + id, + descriptor.format, + descriptor.media_type, + digest, + size_bytes, + ArtifactPayload::new(relative_path), + descriptor.metadata, + descriptor.sources, + descriptor.storage_class, + )) + } + + /// Resolve an artifact's store-relative payload path safely. + pub fn payload_path(&self, artifact: &Artifact) -> Result { + let relative = artifact.payload().relative_path(); + Self::validate_relative_payload_path(relative)?; + Ok(self.root.join(relative)) + } + + /// Return whether the payload referenced by an artifact is present. + pub fn contains(&self, artifact: &Artifact) -> bool { + self.payload_path(artifact) + .map(|path| path.is_file()) + .unwrap_or(false) + } + + /// Open the artifact payload for streaming reads. + pub fn open(&self, artifact: &Artifact) -> Result { + let path = self.payload_path(artifact)?; + File::open(&path) + .with_context(|| format!("Artifact payload '{}' is unavailable", path.display())) + } + + /// Read the complete payload as bytes. + /// + /// Prefer [`open`](Self::open) for large artifacts. + pub fn read_bytes(&self, artifact: &Artifact) -> Result> { + let path = self.payload_path(artifact)?; + fs::read(&path) + .with_context(|| format!("Failed to read artifact payload '{}'", path.display())) + } + + /// Read an artifact through the legacy UTF-8 text compatibility boundary. + pub fn read_text(&self, artifact: &Artifact) -> Result { + let bytes = self.read_bytes(artifact)?; + String::from_utf8(bytes).with_context(|| { + format!( + "Artifact '{}' ({}) is not valid UTF-8; use an artifact-native transform", + artifact.id(), + artifact.media_type() + ) + }) + } + + /// Atomically materialize a stored artifact at its final destination. + /// + /// The destination is replaced only after the complete payload has been + /// copied and synced to a temporary file in the destination directory. + pub fn materialize(&self, artifact: &Artifact, destination: impl AsRef) -> Result<()> { + let destination = destination.as_ref(); + let parent = destination + .parent() + .filter(|path| !path.as_os_str().is_empty()) + .unwrap_or_else(|| Path::new(".")); + fs::create_dir_all(parent).with_context(|| { + format!( + "Failed to create final artifact directory '{}'", + parent.display() + ) + })?; + + let mut input = self.open(artifact)?; + let mut temporary = tempfile::NamedTempFile::new_in(parent).with_context(|| { + format!( + "Failed to create atomic output temporary file in '{}'", + parent.display() + ) + })?; + io::copy(&mut input, &mut temporary).with_context(|| { + format!( + "Failed to copy artifact '{}' to temporary output", + artifact.id() + ) + })?; + temporary.flush().context("Failed to flush final artifact")?; + temporary + .as_file() + .sync_all() + .context("Failed to sync final artifact")?; + temporary + .persist(destination) + .map_err(|error| error.error) + .with_context(|| { + format!( + "Failed to atomically materialize artifact '{}' at '{}'", + artifact.id(), + destination.display() + ) + })?; + Ok(()) + } + + pub(crate) fn temporary_directory(&self) -> PathBuf { + self.root.join(".tmp") + } + + fn object_relative_path(digest_hex: &str) -> PathBuf { + PathBuf::from("objects") + .join("sha256") + .join(&digest_hex[..2]) + .join(digest_hex) + } + + fn validate_relative_payload_path(path: &Path) -> Result<()> { + if path.is_absolute() { + anyhow::bail!("artifact payload path must be store-relative"); + } + for component in path.components() { + match component { + Component::ParentDir | Component::RootDir | Component::Prefix(_) => { + anyhow::bail!("artifact payload path escapes the artifact store") + } + Component::CurDir | Component::Normal(_) => {} + } + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::artifact::ArtifactStorageClass; + use crate::graph::Format; + + fn fixture(format: Format, bytes: &[u8]) { + let directory = tempfile::tempdir().unwrap(); + let store = ArtifactStore::new(directory.path()).unwrap(); + let artifact = store + .put_bytes( + bytes, + ArtifactDescriptor::for_format(format, ArtifactStorageClass::Source), + ) + .unwrap(); + assert_eq!(artifact.size_bytes(), bytes.len() as u64); + assert_eq!(artifact.digest().value().len(), 64); + assert_eq!(store.read_bytes(&artifact).unwrap(), bytes); + } + + #[test] + fn representative_artifact_families_are_binary_safe() { + fixture(Format::Pdf, b"%PDF-1.7\n%\x80\x81\x82\n"); + fixture(Format::Png, b"\x89PNG\r\n\x1a\n\x00\xff\x00"); + fixture(Format::Wav, b"RIFF\x00\x00\x00\x00WAVEfmt \x00\xff"); + fixture(Format::Markdown, b"# artifact kernel\n"); + } + + #[test] + fn identical_payloads_share_content_storage() { + let directory = tempfile::tempdir().unwrap(); + let store = ArtifactStore::new(directory.path()).unwrap(); + let descriptor = || { + ArtifactDescriptor::for_format(Format::Png, ArtifactStorageClass::Intermediate) + }; + let first = store.put_bytes(b"\x89PNG\x00", descriptor()).unwrap(); + let second = store.put_bytes(b"\x89PNG\x00", descriptor()).unwrap(); + assert_eq!(first.id(), second.id()); + assert_eq!(first.digest(), second.digest()); + assert_eq!(first.payload(), second.payload()); + } + + #[test] + fn importing_a_source_never_mutates_the_source_file() { + let directory = tempfile::tempdir().unwrap(); + let source = directory.path().join("source.bin"); + fs::write(&source, [0_u8, 255, 1, 2]).unwrap(); + let store = ArtifactStore::new(directory.path().join("store")).unwrap(); + let artifact = store + .import_path( + &source, + ArtifactDescriptor::for_format(Format::Png, ArtifactStorageClass::Source), + ) + .unwrap(); + assert_eq!(fs::read(&source).unwrap(), vec![0, 255, 1, 2]); + assert_ne!(store.payload_path(&artifact).unwrap(), source); + } + + #[test] + fn materialize_replaces_final_output_after_complete_copy() { + let directory = tempfile::tempdir().unwrap(); + let store = ArtifactStore::new(directory.path().join("store")).unwrap(); + let artifact = store + .put_bytes( + b"new bytes", + ArtifactDescriptor::for_format(Format::Pdf, ArtifactStorageClass::Terminal), + ) + .unwrap(); + let destination = directory.path().join("release.pdf"); + fs::write(&destination, b"old bytes").unwrap(); + store.materialize(&artifact, &destination).unwrap(); + assert_eq!(fs::read(destination).unwrap(), b"new bytes"); + } +} diff --git a/crates/renderflow-core/src/artifact/transform.rs b/crates/renderflow-core/src/artifact/transform.rs new file mode 100644 index 0000000..b803e02 --- /dev/null +++ b/crates/renderflow-core/src/artifact/transform.rs @@ -0,0 +1,99 @@ +use std::sync::Arc; + +use anyhow::{Context, Result}; + +use super::{Artifact, ArtifactDescriptor, ArtifactStorageClass, ArtifactStore}; +use crate::graph::Format; +use crate::transforms::Transform; + +/// Experimental artifact-native transform seam introduced by the artifact kernel. +/// +/// This trait is deliberately small. Issue #357 owns stabilization/versioning of +/// the Transform v2 and plugin SDK contracts. Implementations may inspect or +/// stream the input payload through [`ArtifactStore`] without converting it to +/// UTF-8. +pub trait ArtifactTransform: Send + Sync { + /// Human-readable transform identifier. + fn name(&self) -> &str { + "ArtifactTransform" + } + + /// Stable material used as part of the artifact DAG cache key. + /// + /// Providers with configuration that affects output should override this or + /// be registered with an explicit identity through the executor. + fn cache_identity(&self) -> String { + self.name().to_string() + } + + /// Produce one artifact in `output_format` from `input`. + fn apply( + &self, + input: &Artifact, + output_format: Format, + store: &ArtifactStore, + ) -> Result; +} + +/// Compatibility adapter that runs the existing UTF-8 [`Transform`] API inside +/// the binary-safe artifact executor. +pub struct TextTransformAdapter { + transform: Arc, + cache_identity: String, +} + +impl TextTransformAdapter { + /// Wrap a text transform using its name as cache identity. + pub fn new(transform: Arc) -> Self { + let cache_identity = transform.name().to_string(); + Self { + transform, + cache_identity, + } + } + + /// Wrap a text transform with explicit configuration-aware cache identity. + pub fn with_identity( + transform: Arc, + cache_identity: impl Into, + ) -> Self { + Self { + transform, + cache_identity: cache_identity.into(), + } + } +} + +impl ArtifactTransform for TextTransformAdapter { + fn name(&self) -> &str { + self.transform.name() + } + + fn cache_identity(&self) -> String { + self.cache_identity.clone() + } + + fn apply( + &self, + input: &Artifact, + output_format: Format, + store: &ArtifactStore, + ) -> Result { + let input_text = store.read_text(input).with_context(|| { + format!( + "Text transform '{}' requires UTF-8 input; register an artifact-native transform for binary payloads", + self.transform.name() + ) + })?; + let output = self + .transform + .apply(input_text) + .with_context(|| format!("Text transform '{}' failed", self.transform.name()))?; + store.put_bytes( + output.as_bytes(), + ArtifactDescriptor::for_format(output_format, ArtifactStorageClass::Intermediate) + .with_source(input.id().clone()) + .with_metadata("renderflow.transform", self.transform.name()), + ) + } +} diff --git a/crates/renderflow-core/src/commands/graph_build.rs b/crates/renderflow-core/src/commands/graph_build.rs index 0803c22..14157df 100644 --- a/crates/renderflow-core/src/commands/graph_build.rs +++ b/crates/renderflow-core/src/commands/graph_build.rs @@ -1,9 +1,9 @@ -use std::fs; use std::path::Path; use anyhow::{Context, Result}; use tracing::{debug, info}; +use crate::artifact::{ArtifactDescriptor, ArtifactStorageClass, ArtifactStore}; use crate::config::load_config_for_graph; use crate::files::ensure_output_dir; use crate::graph::Format; @@ -11,20 +11,6 @@ use crate::optimization::OptimizationMode; use crate::transforms::yaml_loader::build_graph_and_executor_from_yaml; /// Run graph-based execution targeting a single output format. -/// -/// The transform graph is resolved automatically from the `transforms` YAML -/// file referenced in `config_path`. The shortest path (according to -/// `optimization`) from the detected source format to `target` is found, -/// and every intermediate and final format is produced. -/// -/// # Errors -/// -/// Returns an error when: -/// * the config file cannot be read or parsed, -/// * no `transforms` key is present in the config, -/// * `target` is not a recognised format, -/// * `target` is not reachable from the source format, -/// * any transform in the execution plan fails. pub fn run_target( config_path: &str, target: &str, @@ -44,18 +30,6 @@ pub fn run_target( } /// Run graph-based execution targeting all formats reachable from the source. -/// -/// The transform graph is resolved automatically from the `transforms` YAML -/// file referenced in `config_path`. Every format reachable from the source -/// format is produced in dependency order. -/// -/// # Errors -/// -/// Returns an error when: -/// * the config file cannot be read or parsed, -/// * no `transforms` key is present in the config, -/// * no output formats are reachable from the source format, -/// * any transform in the execution plan fails. pub fn run_all( config_path: &str, dry_run: bool, @@ -65,9 +39,6 @@ pub fn run_all( } /// Shared implementation for `run_target` and `run_all`. -/// -/// `explicit_targets` is `Some(vec)` for `--target` mode and `None` for -/// `--all` mode (targets are discovered dynamically from the graph). fn run_impl( config_path: &str, explicit_targets: Option>, @@ -95,7 +66,6 @@ fn run_impl( let opt_mode = optimization.unwrap_or(config.optimization); info!(optimization = %opt_mode, "Using optimization mode"); - // Derive the source format from the config's input field. let source_format: Format = config.input_format().to_string().parse().with_context(|| { format!( "Could not map input format '{}' to a known graph format", @@ -103,11 +73,9 @@ fn run_impl( ) })?; - // Determine which formats to build. let targets: Vec = match explicit_targets { - Some(t) => t, + Some(targets) => targets, None => { - // --all: discover every format reachable from the source. let reachable = graph.reachable_from(source_format); if reachable.is_empty() { anyhow::bail!( @@ -120,7 +88,7 @@ fn run_impl( reachable.len(), reachable .iter() - .map(|f| f.to_string()) + .map(|format| format.to_string()) .collect::>() .join(", ") ); @@ -128,7 +96,6 @@ fn run_impl( } }; - // Build the minimal DAG that covers all targets. let dag = graph .build_multi_target_dag_with_mode(source_format, &targets, opt_mode) .ok_or_else(|| { @@ -139,12 +106,11 @@ fn run_impl( ) })?; - // Emit the execution plan when debug logging is enabled. debug!("Execution plan (DAG tree):\n{}", dag.to_tree(source_format)); let input_stem = Path::new(&config.input) .file_stem() - .and_then(|s| s.to_str()) + .and_then(|stem| stem.to_str()) .unwrap_or("document"); let output_dir = if dry_run { @@ -166,26 +132,54 @@ fn run_impl( ensure_output_dir(&config.output_dir)? }; - let executor = executor.with_cache(output_dir.join(".renderflow-dag-cache.json")); - - // Read and execute. - let content = fs::read_to_string(&config.input) - .with_context(|| format!("Failed to read input file: {}", config.input))?; - - info!("Executing graph-based pipeline"); + // Keep intermediate/cache state outside the final output directory itself. + let state_parent = output_dir + .parent() + .filter(|path| !path.as_os_str().is_empty()) + .unwrap_or_else(|| Path::new(".")); + let state_dir = state_parent.join(".renderflow"); + let artifact_store = ArtifactStore::new(state_dir.join("artifacts"))?; + let executor = executor.with_cache(state_dir.join("dag-cache.json")); + + let source_artifact = artifact_store.import_path( + &config.input, + ArtifactDescriptor::for_format(source_format, ArtifactStorageClass::Source), + )?; + + info!( + artifact = %source_artifact.id(), + digest = %source_artifact.digest(), + bytes = source_artifact.size_bytes(), + "Executing graph-based pipeline from binary-safe source artifact" + ); let results = executor - .execute(&dag, source_format, content) + .execute_artifact(&dag, source_format, source_artifact, &artifact_store) .context("Graph execution failed")?; - // Write each produced format to disk (skip the source format). - for (format, output_content) in &results { + for (format, artifact) in &results { if *format == source_format { continue; } let output_path = output_dir.join(format!("{}.{}", input_stem, format)); - fs::write(&output_path, output_content) - .with_context(|| format!("Failed to write output to '{}'", output_path.display()))?; - info!("✔ Output written to: {}", output_path.display()); + let terminal_artifact = artifact + .clone() + .with_storage_class(ArtifactStorageClass::Terminal); + artifact_store + .materialize(&terminal_artifact, &output_path) + .with_context(|| { + format!( + "Failed to materialize '{}' output to '{}'", + format, + output_path.display() + ) + })?; + info!( + artifact = %terminal_artifact.id(), + digest = %terminal_artifact.digest(), + bytes = terminal_artifact.size_bytes(), + "✔ Output written to: {}", + output_path.display() + ); } Ok(()) diff --git a/crates/renderflow-core/src/graph/dag_executor.rs b/crates/renderflow-core/src/graph/dag_executor.rs index 889a249..d2d7fc5 100644 --- a/crates/renderflow-core/src/graph/dag_executor.rs +++ b/crates/renderflow-core/src/graph/dag_executor.rs @@ -1,6 +1,5 @@ use std::collections::HashMap; -use std::io::Write; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; use anyhow::{Context, Result}; @@ -8,88 +7,25 @@ use rayon::prelude::*; use tracing::{debug, warn}; use super::{Format, MultiTargetDag, TransformEdge}; -use crate::cache::{compute_dag_node_hash, load_cache, save_cache, TransformCache}; +use crate::artifact::{ + compute_artifact_node_hash, load_artifact_cache, save_artifact_cache, Artifact, + ArtifactCache, ArtifactCollection, ArtifactDescriptor, ArtifactStorageClass, ArtifactStore, + ArtifactTransform, TextTransformAdapter, +}; use crate::transforms::aggregation::AggregationTransform; use crate::transforms::Transform; -/// Executes a [`MultiTargetDag`] in correct dependency order. +/// Executes a [`MultiTargetDag`] using file-backed, binary-safe artifacts. /// -/// Single-input edges ([`InputKind::Single`](super::InputKind::Single)) are -/// dispatched to a registered [`Transform`]; collection edges -/// ([`InputKind::Collection`](super::InputKind::Collection)) are dispatched to -/// a registered [`AggregationTransform`]. -/// -/// Independent edges — those whose source format becomes available at the same -/// time — are grouped into *waves* and executed in parallel using Rayon. -/// -/// # Registration -/// -/// Before calling [`execute`](DagExecutor::execute), register a transform for -/// every edge that appears in the DAG you intend to execute: -/// -/// * **Single-input edges** – call -/// [`register_single`](DagExecutor::register_single) with an -/// `Arc`-wrapped [`Transform`] implementation that is [`Send`] + [`Sync`]. -/// -/// * **Collection edges** – call -/// [`register_aggregation`](DagExecutor::register_aggregation) with an -/// `Arc`-wrapped [`AggregationTransform`]. -/// -/// # Execution -/// -/// [`execute`](DagExecutor::execute) starts with one piece of initial content -/// for the source format and repeatedly processes waves until no more edges can -/// run. Each wave contains all edges whose source format has already been -/// produced. Edges within the same wave are executed in parallel. -/// -/// The returned map contains the produced content for every format that was -/// reached during execution, including the initial source format. -/// -/// # Errors -/// -/// * A transform for a DAG edge has not been registered. -/// * A registered transform returns an error during execution. -/// * Temporary file I/O fails when processing a collection edge. -/// -/// # Example -/// -/// ```rust -/// use renderflow::graph::{DagExecutor, Format, TransformEdge, TransformGraph}; -/// use renderflow::transforms::Transform; -/// use anyhow::Result; -/// use std::sync::Arc; -/// -/// struct UpperTransform; -/// impl Transform for UpperTransform { -/// fn apply(&self, input: String) -> Result { -/// Ok(input.to_uppercase()) -/// } -/// } -/// -/// let mut graph = TransformGraph::new(); -/// graph.add_transform(TransformEdge::new(Format::Markdown, Format::Html, 0.5, 1.0)); -/// let dag = graph -/// .build_multi_target_dag(Format::Markdown, &[Format::Html]) -/// .unwrap(); -/// -/// let mut executor = DagExecutor::new(); -/// executor.register_single( -/// Format::Markdown, -/// Format::Html, -/// Arc::new(UpperTransform), -/// ); -/// -/// let results = executor -/// .execute(&dag, Format::Markdown, "hello".to_string()) -/// .unwrap(); -/// assert_eq!(results[&Format::Html], "HELLO"); -/// ``` +/// The artifact-native executor is the canonical substrate. The legacy +/// [`execute`](Self::execute) method remains as a UTF-8 compatibility wrapper +/// for existing callers and transforms. pub struct DagExecutor { - /// Single-input transforms keyed by `(from, to)` format pair. - single_transforms: HashMap<(Format, Format), Arc>, + /// Single-input artifact transforms keyed by `(from, to)` format pair. + single_transforms: HashMap<(Format, Format), Arc>, /// Collection-input transforms keyed by `(from, to)` format pair. aggregation_transforms: HashMap<(Format, Format), Arc>, - /// Optional path for on-disk caching of single-transform outputs. + /// Optional artifact-native DAG cache path. cache_path: Option, } @@ -103,39 +39,62 @@ impl DagExecutor { } } - /// Configure an on-disk cache for single-transform outputs. - /// - /// When set, [`execute`](DagExecutor::execute) loads the cache from `path` - /// before processing, skips any single-input edge whose output is already - /// cached, and saves the updated cache back to `path` after all waves - /// complete. + /// Configure the artifact-native on-disk DAG cache. /// - /// If the file does not exist the executor starts with an empty cache. - /// Errors loading or saving the cache are logged at `WARN` level and never - /// abort the build. + /// Legacy string-cache files are intentionally treated as cache misses and + /// replaced with the artifact-cache schema on the next successful save. pub fn with_cache(mut self, path: impl Into) -> Self { self.cache_path = Some(path.into()); self } - /// Register a single-input transform for the `from → to` edge. - /// - /// If a transform is already registered for the same `(from, to)` pair it - /// is replaced by the new one. + /// Register an existing UTF-8 text transform through the compatibility adapter. pub fn register_single( &mut self, from: Format, to: Format, transform: Arc, + ) -> &mut Self { + self.single_transforms.insert( + (from, to), + Arc::new(TextTransformAdapter::new(transform)), + ); + self + } + + /// Register a text transform with configuration-aware cache identity. + /// + /// Embedders that know configuration affecting transform output can use this + /// seam while the versioned Transform v2 contract is developed in #357. + pub fn register_single_with_identity( + &mut self, + from: Format, + to: Format, + transform: Arc, + cache_identity: impl Into, + ) -> &mut Self { + self.single_transforms.insert( + (from, to), + Arc::new(TextTransformAdapter::with_identity( + transform, + cache_identity, + )), + ); + self + } + + /// Register an artifact-native transform that may consume arbitrary bytes. + pub fn register_artifact( + &mut self, + from: Format, + to: Format, + transform: Arc, ) -> &mut Self { self.single_transforms.insert((from, to), transform); self } /// Register a collection-input transform for the `from → to` edge. - /// - /// If a transform is already registered for the same `(from, to)` pair it - /// is replaced by the new one. pub fn register_aggregation( &mut self, from: Format, @@ -146,180 +105,255 @@ impl DagExecutor { self } - /// Execute the DAG starting from `initial_content` for `source_format`. - /// - /// Returns a [`HashMap`] mapping every reachable format (including - /// `source_format`) to the string content produced for it. + /// Execute a DAG using the legacy UTF-8 `String` API. /// - /// Edges are processed in topological *waves*. In each wave all edges - /// whose source format is already available are collected and executed in - /// parallel using Rayon. The wave's outputs are added to the available - /// set before the next wave begins. - /// - /// # Errors - /// - /// Returns the first error encountered if any transform fails or if a - /// required transform has not been registered. + /// Existing text transforms continue to work unchanged, but callers that + /// expect binary output must use [`execute_artifact`](Self::execute_artifact) + /// or [`execute_artifacts`](Self::execute_artifacts). pub fn execute( &self, dag: &MultiTargetDag, source_format: Format, initial_content: String, ) -> Result> { - // Load an on-disk cache when a cache path is configured. - let cache: Option> = self + let temporary_directory = if self.cache_path.is_none() { + Some(tempfile::tempdir().context("Failed to create legacy DAG work directory")?) + } else { + None + }; + + let store_root = if let Some(cache_path) = &self.cache_path { + Self::legacy_store_path(cache_path) + } else { + temporary_directory + .as_ref() + .expect("temporary directory exists when no cache is configured") + .path() + .join("artifacts") + }; + let store = ArtifactStore::new(store_root)?; + let source = store.put_bytes( + initial_content.as_bytes(), + ArtifactDescriptor::for_format(source_format, ArtifactStorageClass::Source), + )?; + let artifacts = self.execute_artifact(dag, source_format, source, &store)?; + + artifacts + .into_iter() + .map(|(format, artifact)| { + let text = store.read_text(&artifact).with_context(|| { + format!( + "Legacy String DAG API cannot return binary '{}' output; use execute_artifact", + format + ) + })?; + Ok((format, text)) + }) + .collect() + } + + /// Execute a DAG from one binary-safe source artifact. + pub fn execute_artifact( + &self, + dag: &MultiTargetDag, + source_format: Format, + initial_artifact: Artifact, + store: &ArtifactStore, + ) -> Result> { + let collections = self.execute_artifacts( + dag, + source_format, + ArtifactCollection::one(initial_artifact), + store, + )?; + + collections + .into_iter() + .map(|(format, collection)| { + let artifact = collection.into_one().with_context(|| { + format!( + "Format '{}' produced a collection where a single artifact was expected", + format + ) + })?; + Ok((format, artifact)) + }) + .collect() + } + + /// Execute a DAG from an ordered source artifact collection. + /// + /// Single-input edges require exactly one artifact. Collection edges receive + /// every artifact in declared order as file-backed paths and may therefore + /// aggregate binary inputs without converting them to text. + pub fn execute_artifacts( + &self, + dag: &MultiTargetDag, + source_format: Format, + initial_artifacts: ArtifactCollection, + store: &ArtifactStore, + ) -> Result> { + if initial_artifacts.is_empty() { + anyhow::bail!("Artifact DAG execution requires at least one source artifact"); + } + + let cache: Option> = self .cache_path .as_deref() - .map(|p| Mutex::new(load_cache(p))); - - // Track the content produced for every format encountered so far. - let mut available: HashMap = HashMap::new(); - available.insert(source_format, initial_content); + .map(|path| Mutex::new(load_artifact_cache(path))); + let mut available: HashMap = HashMap::new(); + available.insert(source_format, initial_artifacts); let mut remaining: Vec<&TransformEdge> = dag.execution_order(); loop { - // Partition: edges ready to execute vs. those still waiting. let (wave, next_remaining): (Vec<_>, Vec<_>) = remaining .into_iter() - .partition(|e| available.contains_key(&e.from)); + .partition(|edge| available.contains_key(&edge.from)); if wave.is_empty() { if !next_remaining.is_empty() { warn!( unreachable = next_remaining.len(), - "Some DAG edges could not be executed because their \ - source format was never produced" + "Some DAG edges could not execute because their source format was never produced" ); } break; } - debug!(wave_size = wave.len(), "Executing DAG wave"); - - // Execute all edges in the current wave in parallel. - let wave_results: Result> = wave + debug!(wave_size = wave.len(), "Executing artifact DAG wave"); + let wave_results: Result> = wave .into_par_iter() - .map(|edge| self.execute_edge(edge, &available, cache.as_ref())) + .map(|edge| self.execute_edge(edge, &available, store, cache.as_ref())) .collect(); - // Propagate the first error; otherwise record the new outputs. - for (format, content) in wave_results? { - available.insert(format, content); + for (format, artifacts) in wave_results? { + available.insert(format, artifacts); } - remaining = next_remaining; } - // Persist the updated cache to disk when configured. - if let Some(path) = &self.cache_path { - if let Some(cache_mutex) = cache { - match cache_mutex.into_inner() { - Ok(cache) => { - if let Err(e) = save_cache(&cache, path) { - warn!(error = %e, path = %path.display(), "Failed to save DAG cache"); - } - } - Err(e) => { - warn!(error = %e, "DAG cache mutex was poisoned; cache not saved"); + if let (Some(cache_path), Some(cache_mutex)) = (&self.cache_path, cache) { + match cache_mutex.into_inner() { + Ok(cache) => { + if let Err(error) = save_artifact_cache(&cache, cache_path) { + warn!( + error = %error, + path = %cache_path.display(), + "Failed to save artifact DAG cache" + ); } } + Err(error) => { + warn!( + error = %error, + "Artifact DAG cache mutex was poisoned; cache not saved" + ); + } } } Ok(available) } - // ── private helpers ─────────────────────────────────────────────────────── - - /// Dispatch a single edge to the appropriate executor. fn execute_edge( &self, edge: &TransformEdge, - available: &HashMap, - cache: Option<&Mutex>, - ) -> Result<(Format, String)> { - let input = available - .get(&edge.from) - .expect("source format must be in available set when execute_edge is called") - .clone(); + available: &HashMap, + store: &ArtifactStore, + cache: Option<&Mutex>, + ) -> Result<(Format, ArtifactCollection)> { + let inputs = available.get(&edge.from).ok_or_else(|| { + anyhow::anyhow!( + "Source format '{}' was not available for DAG edge", + edge.from + ) + })?; if edge.input_kind.is_single() { - self.execute_single_edge(edge, input, cache) + let input = inputs.clone().into_one().with_context(|| { + format!( + "Single transform {:?} → {:?} requires exactly one artifact", + edge.from, edge.to + ) + })?; + let output = self.execute_single_edge(edge, &input, store, cache)?; + Ok((edge.to, ArtifactCollection::one(output))) } else { - self.execute_collection_edge(edge, &[input]) + let output = self.execute_collection_edge(edge, inputs, store)?; + Ok((edge.to, ArtifactCollection::one(output))) } } - /// Execute a single-input edge by delegating to the registered transform. fn execute_single_edge( &self, edge: &TransformEdge, - input: String, - cache: Option<&Mutex>, - ) -> Result<(Format, String)> { - let from_str = edge.from.to_string(); - let to_str = edge.to.to_string(); - - // Check the cache before running the transform. - if let Some(cache_mutex) = cache { - let hash = compute_dag_node_hash(&input, &from_str, &to_str); - - if let Ok(guard) = cache_mutex.lock() { - if let Some(cached_output) = guard.get(&hash) { - debug!(from = ?edge.from, to = ?edge.to, "Cache hit; skipping transform"); - return Ok((edge.to, cached_output.to_string())); - } - } - - // Cache miss: run the transform and store the result. - let output = self.run_single_transform(edge, input)?; - - if let Ok(mut guard) = cache_mutex.lock() { - guard.insert(hash, output.clone()); - } - - return Ok((edge.to, output)); - } - - // No cache configured: run the transform directly. - let output = self.run_single_transform(edge, input)?; - Ok((edge.to, output)) - } - - /// Look up and apply the registered single-input transform for `edge`. - fn run_single_transform(&self, edge: &TransformEdge, input: String) -> Result { + input: &Artifact, + store: &ArtifactStore, + cache: Option<&Mutex>, + ) -> Result { let transform = self .single_transforms .get(&(edge.from, edge.to)) .ok_or_else(|| { anyhow::anyhow!( - "No single transform registered for {:?} → {:?}", + "No artifact transform registered for {:?} → {:?}", edge.from, edge.to ) })?; + let cache_identity = transform.cache_identity(); + let cache_key = + compute_artifact_node_hash(input, edge.from, edge.to, &cache_identity); - debug!(from = ?edge.from, to = ?edge.to, "Executing single transform"); + if let Some(cache_mutex) = cache { + if let Ok(guard) = cache_mutex.lock() { + if let Some(cached) = guard.get(&cache_key) { + if store.contains(cached) { + debug!( + from = ?edge.from, + to = ?edge.to, + artifact = %cached.id(), + "Artifact cache hit; skipping transform" + ); + return Ok(cached + .clone() + .with_storage_class(ArtifactStorageClass::Cached)); + } + } + } + } - let output = transform - .apply(input) - .with_context(|| format!("Single transform {:?} → {:?} failed", edge.from, edge.to))?; + debug!( + from = ?edge.from, + to = ?edge.to, + transform = %transform.name(), + "Executing artifact transform" + ); + let output = transform.apply(input, edge.to, store).with_context(|| { + format!( + "Artifact transform {:?} → {:?} ({}) failed", + edge.from, + edge.to, + transform.name() + ) + })?; + self.validate_output_format(edge, &output)?; - debug!(from = ?edge.from, to = ?edge.to, "Single transform completed"); + if let Some(cache_mutex) = cache { + if let Ok(mut guard) = cache_mutex.lock() { + guard.insert(cache_key, output.clone()); + } + } Ok(output) } - /// Execute a collection-input edge. - /// - /// Each input string is written to a temporary file so that - /// [`AggregationTransform::aggregate`] receives file paths, as its - /// contract requires. The aggregated output is read back as a `String`. fn execute_collection_edge( &self, edge: &TransformEdge, - inputs: &[String], - ) -> Result<(Format, String)> { + inputs: &ArtifactCollection, + store: &ArtifactStore, + ) -> Result { let transform = self .aggregation_transforms .get(&(edge.from, edge.to)) @@ -331,60 +365,84 @@ impl DagExecutor { ) })?; - debug!( - from = ?edge.from, - to = ?edge.to, - inputs = inputs.len(), - "Executing collection transform" - ); - - // Write each input to a temporary file so the aggregation transform - // receives proper file paths. - let temp_inputs: Vec = inputs + let input_paths: Vec = inputs .iter() - .map(|content| { - let mut f = tempfile::NamedTempFile::new() - .context("Failed to create temp file for aggregation input")?; - f.write_all(content.as_bytes()) - .context("Failed to write to aggregation input temp file")?; - Ok(f) - }) + .map(|artifact| store.payload_path(artifact)) .collect::>>()?; - - let input_paths: Vec<&str> = temp_inputs + let input_path_strings: Vec<&str> = input_paths .iter() - .map(|f| { - f.path().to_str().ok_or_else(|| { - anyhow::anyhow!("Aggregation input temp file path is not valid UTF-8") + .map(|path| { + path.to_str().ok_or_else(|| { + anyhow::anyhow!( + "Artifact-store path '{}' is not valid UTF-8", + path.display() + ) }) }) .collect::>>()?; - // Create a temp file to receive the aggregated output. - let temp_output = tempfile::NamedTempFile::new() - .context("Failed to create temp file for aggregation output")?; - let output_path = temp_output.path().to_str().ok_or_else(|| { - anyhow::anyhow!("Aggregation output temp file path is not valid UTF-8") - })?; + let suffix = format!(".{}", edge.to); + let temporary_output = tempfile::Builder::new() + .prefix("aggregate-") + .suffix(&suffix) + .tempfile_in(store.temporary_directory()) + .context("Failed to create aggregation output temporary file")? + .into_temp_path(); + let output_path = temporary_output + .to_str() + .ok_or_else(|| anyhow::anyhow!("Aggregation output path is not valid UTF-8"))?; + debug!( + from = ?edge.from, + to = ?edge.to, + transform = %transform.name(), + inputs = inputs.len(), + "Executing artifact collection transform" + ); transform - .aggregate(&input_paths, output_path) + .aggregate(&input_path_strings, output_path) .with_context(|| { format!( - "Collection transform {:?} → {:?} failed", - edge.from, edge.to + "Collection transform {:?} → {:?} ({}) failed", + edge.from, + edge.to, + transform.name() ) })?; - let output = std::fs::read_to_string(temp_output.path()).with_context(|| { - format!( - "Failed to read aggregation output for {:?} → {:?}", - edge.from, edge.to - ) - })?; + let sources = inputs.iter().map(|artifact| artifact.id().clone()); + let output = store.import_path( + &temporary_output, + ArtifactDescriptor::for_format(edge.to, ArtifactStorageClass::Intermediate) + .with_sources(sources) + .with_metadata("renderflow.transform", transform.name()), + )?; + self.validate_output_format(edge, &output)?; + Ok(output) + } + + fn validate_output_format(&self, edge: &TransformEdge, output: &Artifact) -> Result<()> { + if output.format().as_str() != edge.to.to_string() { + anyhow::bail!( + "Transform {:?} → {:?} returned artifact format '{}'", + edge.from, + edge.to, + output.format() + ); + } + Ok(()) + } - debug!(from = ?edge.from, to = ?edge.to, "Collection transform completed"); - Ok((edge.to, output)) + fn legacy_store_path(cache_path: &Path) -> PathBuf { + let parent = cache_path + .parent() + .filter(|candidate| !candidate.as_os_str().is_empty()) + .unwrap_or_else(|| Path::new(".")); + let stem = cache_path + .file_stem() + .and_then(|value| value.to_str()) + .unwrap_or("dag-cache"); + parent.join(format!(".{}-artifacts", stem)) } } @@ -396,28 +454,19 @@ impl Default for DagExecutor { #[cfg(test)] mod tests { - use super::*; - use crate::graph::{Format, InputKind, TransformEdge, TransformGraph}; - use anyhow::{bail, Result}; - use std::sync::Arc; + use std::io::Write; + use std::sync::atomic::{AtomicUsize, Ordering}; - // ── helpers ─────────────────────────────────────────────────────────────── + use anyhow::{bail, Result}; - /// Build a small graph: Markdown → Html → {Pdf, Docx} - fn build_graph() -> TransformGraph { - let mut g = TransformGraph::new(); - g.add_transform(TransformEdge::new(Format::Markdown, Format::Html, 0.5, 1.0)); - g.add_transform(TransformEdge::new(Format::Html, Format::Pdf, 0.8, 0.85)); - g.add_transform(TransformEdge::new(Format::Html, Format::Docx, 0.6, 0.90)); - g - } + use super::*; + use crate::graph::{InputKind, TransformGraph}; - /// A [`Transform`] that appends a fixed string to its input. - struct AppendTransform(String); + struct AppendTransform(&'static str); impl Transform for AppendTransform { fn name(&self) -> &str { - "AppendTransform" + "append" } fn apply(&self, input: String) -> Result { @@ -425,554 +474,268 @@ mod tests { } } - /// A [`Transform`] that always returns an error. - struct FailingTransform; + struct BinaryCopyTransform; - impl Transform for FailingTransform { + impl ArtifactTransform for BinaryCopyTransform { fn name(&self) -> &str { - "FailingTransform" + "binary-copy" } - fn apply(&self, _input: String) -> Result { - bail!("intentional transform failure") + fn apply( + &self, + input: &Artifact, + output_format: Format, + store: &ArtifactStore, + ) -> Result { + let mut reader = store.open(input)?; + store.put_reader( + &mut reader, + ArtifactDescriptor::for_format( + output_format, + ArtifactStorageClass::Intermediate, + ) + .with_source(input.id().clone()), + ) } } - /// An [`AggregationTransform`] that writes all inputs joined by commas. - struct JoinAggregation; + struct CountingBinaryTransform { + executions: Arc, + } - impl AggregationTransform for JoinAggregation { + impl ArtifactTransform for CountingBinaryTransform { fn name(&self) -> &str { - "join" + "counting-binary" } - fn aggregate(&self, inputs: &[&str], output_path: &str) -> Result<()> { - // Read each input file and join the contents. - let parts: Result> = inputs - .iter() - .map(|p| std::fs::read_to_string(p).context("Failed to read aggregation input")) - .collect(); - std::fs::write(output_path, parts?.join(","))?; - Ok(()) + fn cache_identity(&self) -> String { + "counting-binary:v1".to_string() } - } - fn arc_single(t: T) -> Arc { - Arc::new(t) + fn apply( + &self, + input: &Artifact, + output_format: Format, + store: &ArtifactStore, + ) -> Result { + self.executions.fetch_add(1, Ordering::SeqCst); + let mut reader = store.open(input)?; + store.put_reader( + &mut reader, + ArtifactDescriptor::for_format( + output_format, + ArtifactStorageClass::Intermediate, + ) + .with_source(input.id().clone()), + ) + } } - fn arc_agg(t: T) -> Arc { - Arc::new(t) - } + struct OrderedJoinAggregation; - // ── basic single transform ──────────────────────────────────────────────── + impl AggregationTransform for OrderedJoinAggregation { + fn name(&self) -> &str { + "ordered-join" + } - #[test] - fn test_execute_single_transform_produces_output() { - let dag = build_graph() - .build_multi_target_dag(Format::Markdown, &[Format::Html]) - .unwrap(); + fn aggregate(&self, inputs: &[&str], output_path: &str) -> Result<()> { + let mut output = std::fs::File::create(output_path)?; + for path in inputs { + output.write_all(&std::fs::read(path)?)?; + } + Ok(()) + } + } - let mut executor = DagExecutor::new(); - executor.register_single( - Format::Markdown, - Format::Html, - arc_single(AppendTransform("→html".to_string())), - ); + struct FailingAggregation; - let results = executor - .execute(&dag, Format::Markdown, "content".to_string()) - .unwrap(); + impl AggregationTransform for FailingAggregation { + fn name(&self) -> &str { + "failing-aggregation" + } - assert_eq!(results[&Format::Html], "content→html"); + fn aggregate(&self, _inputs: &[&str], output_path: &str) -> Result<()> { + std::fs::write(output_path, b"partial")?; + bail!("intentional failure") + } } - #[test] - fn test_execute_includes_source_format_in_results() { - let dag = build_graph() - .build_multi_target_dag(Format::Markdown, &[Format::Html]) - .unwrap(); + fn one_edge(from: Format, to: Format, input_kind: InputKind) -> MultiTargetDag { + let mut graph = TransformGraph::new(); + graph.add_transform(TransformEdge::with_input_kind( + from, + to, + 1.0, + 1.0, + input_kind, + )); + graph + .build_multi_target_dag(from, &[to]) + .expect("edge must be reachable") + } + #[test] + fn legacy_text_api_runs_through_artifact_adapter() { + let dag = one_edge(Format::Markdown, Format::Html, InputKind::Single); let mut executor = DagExecutor::new(); executor.register_single( Format::Markdown, Format::Html, - arc_single(AppendTransform("→html".to_string())), + Arc::new(AppendTransform("!")), ); - let results = executor - .execute(&dag, Format::Markdown, "source".to_string()) + .execute(&dag, Format::Markdown, "hello".to_string()) .unwrap(); - - assert!( - results.contains_key(&Format::Markdown), - "source format must be present in results" - ); - assert_eq!(results[&Format::Markdown], "source"); + assert_eq!(results[&Format::Html], "hello!"); } - // ── multi-level DAG (dependency order) ──────────────────────────────────── - #[test] - fn test_execute_multi_level_respects_dependency_order() { - let dag = build_graph() - .build_multi_target_dag(Format::Markdown, &[Format::Pdf]) - .unwrap(); - - let mut executor = DagExecutor::new(); - executor - .register_single( - Format::Markdown, - Format::Html, - arc_single(AppendTransform("→html".to_string())), + fn binary_artifact_traverses_graph_without_utf8_conversion() { + let dag = one_edge(Format::Png, Format::Webp, InputKind::Single); + let directory = tempfile::tempdir().unwrap(); + let store = ArtifactStore::new(directory.path().join("store")).unwrap(); + let input_bytes = [0_u8, 159, 255, 1, 2, 3]; + let source = store + .put_bytes( + &input_bytes, + ArtifactDescriptor::for_format(Format::Png, ArtifactStorageClass::Source), ) - .register_single( - Format::Html, - Format::Pdf, - arc_single(AppendTransform("→pdf".to_string())), - ); - - let results = executor - .execute(&dag, Format::Markdown, "start".to_string()) - .unwrap(); - - // Html must have been produced from Markdown before Pdf was produced from Html. - assert_eq!(results[&Format::Html], "start→html"); - assert_eq!(results[&Format::Pdf], "start→html→pdf"); - } - - // ── parallel wave (independent edges) ──────────────────────────────────── - - #[test] - fn test_execute_independent_edges_both_produced() { - let dag = build_graph() - .build_multi_target_dag(Format::Markdown, &[Format::Pdf, Format::Docx]) .unwrap(); - let mut executor = DagExecutor::new(); - executor - .register_single( - Format::Markdown, - Format::Html, - arc_single(AppendTransform("→html".to_string())), - ) - .register_single( - Format::Html, - Format::Pdf, - arc_single(AppendTransform("→pdf".to_string())), - ) - .register_single( - Format::Html, - Format::Docx, - arc_single(AppendTransform("→docx".to_string())), - ); - - let results = executor - .execute(&dag, Format::Markdown, "start".to_string()) - .unwrap(); - - assert_eq!(results[&Format::Html], "start→html"); - assert_eq!(results[&Format::Pdf], "start→html→pdf"); - assert_eq!(results[&Format::Docx], "start→html→docx"); - } - - // ── empty DAG ───────────────────────────────────────────────────────────── - - #[test] - fn test_execute_empty_dag_returns_only_source() { - let dag = build_graph() - .build_multi_target_dag(Format::Markdown, &[]) - .unwrap(); - - let executor = DagExecutor::new(); - let results = executor - .execute(&dag, Format::Markdown, "content".to_string()) - .unwrap(); - - assert_eq!(results.len(), 1); - assert_eq!(results[&Format::Markdown], "content"); - } - - // ── missing transform → error ───────────────────────────────────────────── - - #[test] - fn test_execute_missing_single_transform_returns_error() { - let dag = build_graph() - .build_multi_target_dag(Format::Markdown, &[Format::Html]) - .unwrap(); - - let executor = DagExecutor::new(); // no transforms registered - - let err = executor - .execute(&dag, Format::Markdown, "content".to_string()) - .unwrap_err(); - - assert!( - err.to_string().contains("No single transform registered"), - "error message: {}", - err + executor.register_artifact( + Format::Png, + Format::Webp, + Arc::new(BinaryCopyTransform), ); - } - #[test] - fn test_execute_missing_aggregation_transform_returns_error() { - let mut g = TransformGraph::new(); - g.add_transform(TransformEdge::with_input_kind( - Format::Markdown, - Format::Epub, - 1.0, - 0.85, - InputKind::Collection, - )); - let dag = g - .build_multi_target_dag(Format::Markdown, &[Format::Epub]) + let results = executor + .execute_artifact(&dag, Format::Png, source.clone(), &store) .unwrap(); - - let executor = DagExecutor::new(); // no aggregation registered - - let err = executor - .execute(&dag, Format::Markdown, "content".to_string()) - .unwrap_err(); - - assert!( - err.to_string() - .contains("No aggregation transform registered"), - "error message: {}", - err - ); + let output = &results[&Format::Webp]; + assert_eq!(store.read_bytes(output).unwrap(), input_bytes); + assert_eq!(output.sources(), &[source.id().clone()]); + assert_ne!(output.id(), source.id()); } - // ── failing transform → error propagated ───────────────────────────────── - #[test] - fn test_execute_failing_transform_returns_error() { - let dag = build_graph() - .build_multi_target_dag(Format::Markdown, &[Format::Html]) + fn ordered_collections_are_first_class_aggregation_inputs() { + let dag = one_edge(Format::Png, Format::Pdf, InputKind::Collection); + let directory = tempfile::tempdir().unwrap(); + let store = ArtifactStore::new(directory.path().join("store")).unwrap(); + let first = store + .put_bytes( + b"page-one|", + ArtifactDescriptor::for_format(Format::Png, ArtifactStorageClass::Source), + ) .unwrap(); - - let mut executor = DagExecutor::new(); - executor.register_single(Format::Markdown, Format::Html, arc_single(FailingTransform)); - - let err = executor - .execute(&dag, Format::Markdown, "content".to_string()) - .unwrap_err(); - - // The full error chain (anyhow's alternate format) must contain the - // root cause message emitted by FailingTransform. - let chain = format!("{:#}", err); - assert!( - chain.contains("intentional transform failure"), - "error chain: {}", - chain - ); - } - - #[test] - fn test_execute_error_context_identifies_edge() { - let dag = build_graph() - .build_multi_target_dag(Format::Markdown, &[Format::Html]) + let second = store + .put_bytes( + b"page-two", + ArtifactDescriptor::for_format(Format::Png, ArtifactStorageClass::Source), + ) .unwrap(); - let mut executor = DagExecutor::new(); - executor.register_single(Format::Markdown, Format::Html, arc_single(FailingTransform)); - - let err = executor - .execute(&dag, Format::Markdown, "content".to_string()) - .unwrap_err(); - - // The error chain must identify the failing edge. - let chain = format!("{:#}", err); - assert!( - chain.contains("Markdown") || chain.contains("Html"), - "error chain should identify the edge, got: {chain}" + executor.register_aggregation( + Format::Png, + Format::Pdf, + Arc::new(OrderedJoinAggregation), ); - } - - // ── collection edge ─────────────────────────────────────────────────────── - - #[test] - fn test_execute_collection_transform_produces_output() { - let mut g = TransformGraph::new(); - g.add_transform(TransformEdge::with_input_kind( - Format::Markdown, - Format::Epub, - 1.0, - 0.85, - InputKind::Collection, - )); - let dag = g - .build_multi_target_dag(Format::Markdown, &[Format::Epub]) - .unwrap(); - - let mut executor = DagExecutor::new(); - executor.register_aggregation(Format::Markdown, Format::Epub, arc_agg(JoinAggregation)); let results = executor - .execute(&dag, Format::Markdown, "page content".to_string()) - .unwrap(); - - assert!(results.contains_key(&Format::Epub)); - assert_eq!(results[&Format::Epub], "page content"); - } - - // ── register_single replaces existing ───────────────────────────────────── - - #[test] - fn test_register_single_replaces_previous() { - let dag = build_graph() - .build_multi_target_dag(Format::Markdown, &[Format::Html]) - .unwrap(); - - let mut executor = DagExecutor::new(); - executor - .register_single( - Format::Markdown, - Format::Html, - arc_single(AppendTransform("→old".to_string())), + .execute_artifacts( + &dag, + Format::Png, + ArtifactCollection::new(vec![first.clone(), second.clone()]), + &store, ) - .register_single( - Format::Markdown, - Format::Html, - arc_single(AppendTransform("→new".to_string())), - ); - - let results = executor - .execute(&dag, Format::Markdown, "x".to_string()) .unwrap(); - + let output = results[&Format::Pdf].clone().into_one().unwrap(); + assert_eq!(store.read_bytes(&output).unwrap(), b"page-one|page-two"); assert_eq!( - results[&Format::Html], - "x→new", - "second registration must win" + output.sources(), + &[first.id().clone(), second.id().clone()] ); } - // ── register_aggregation replaces existing ──────────────────────────────── - #[test] - fn test_register_aggregation_replaces_previous() { - let mut g = TransformGraph::new(); - g.add_transform(TransformEdge::with_input_kind( - Format::Markdown, - Format::Epub, - 1.0, - 0.85, - InputKind::Collection, - )); - let dag = g - .build_multi_target_dag(Format::Markdown, &[Format::Epub]) - .unwrap(); - - struct PrefixAggregation(&'static str); - impl AggregationTransform for PrefixAggregation { - fn name(&self) -> &str { - self.0 - } - fn aggregate(&self, _inputs: &[&str], output_path: &str) -> Result<()> { - std::fs::write(output_path, self.0)?; - Ok(()) - } - } - - let mut executor = DagExecutor::new(); - executor - .register_aggregation( - Format::Markdown, - Format::Epub, - arc_agg(PrefixAggregation("first")), + fn artifact_cache_reuses_stored_payload_by_artifact_identity() { + let dag = one_edge(Format::Png, Format::Webp, InputKind::Single); + let directory = tempfile::tempdir().unwrap(); + let store = ArtifactStore::new(directory.path().join("store")).unwrap(); + let cache_path = directory.path().join("dag-cache.json"); + let source = store + .put_bytes( + &[0, 255, 4, 5], + ArtifactDescriptor::for_format(Format::Png, ArtifactStorageClass::Source), ) - .register_aggregation( - Format::Markdown, - Format::Epub, - arc_agg(PrefixAggregation("second")), - ); - - let results = executor - .execute(&dag, Format::Markdown, "input".to_string()) .unwrap(); - - assert_eq!( - results[&Format::Epub], - "second", - "second registration must win" + let executions = Arc::new(AtomicUsize::new(0)); + let mut executor = DagExecutor::new().with_cache(&cache_path); + executor.register_artifact( + Format::Png, + Format::Webp, + Arc::new(CountingBinaryTransform { + executions: Arc::clone(&executions), + }), ); - } - - // ── default ─────────────────────────────────────────────────────────────── - - #[test] - fn test_default_is_empty() { - let executor = DagExecutor::default(); - let dag = build_graph() - .build_multi_target_dag(Format::Markdown, &[]) - .unwrap(); - let results = executor - .execute(&dag, Format::Markdown, "content".to_string()) - .unwrap(); - - // Default executor, empty DAG: just the source. - assert_eq!(results.len(), 1); - } - - // ── caching ─────────────────────────────────────────────────────────────── - - /// A [`Transform`] that counts how many times it has been called. - struct CountingTransform(std::sync::atomic::AtomicUsize); - - impl CountingTransform { - fn new() -> Self { - Self(std::sync::atomic::AtomicUsize::new(0)) - } - - fn call_count(&self) -> usize { - self.0.load(std::sync::atomic::Ordering::SeqCst) - } - } - - impl Transform for CountingTransform { - fn name(&self) -> &str { - "CountingTransform" - } - - fn apply(&self, input: String) -> Result { - self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst); - Ok(format!("{input}→counted")) - } - } - - #[test] - fn test_with_cache_produces_correct_output() { - let dir = tempfile::tempdir().unwrap(); - let cache_file = dir.path().join("dag-cache.json"); - - let dag = build_graph() - .build_multi_target_dag(Format::Markdown, &[Format::Html]) + executor + .execute_artifact(&dag, Format::Png, source.clone(), &store) .unwrap(); - - let mut executor = DagExecutor::new().with_cache(&cache_file); - executor.register_single( - Format::Markdown, - Format::Html, - arc_single(AppendTransform("→html".to_string())), - ); - - let results = executor - .execute(&dag, Format::Markdown, "content".to_string()) + executor + .execute_artifact(&dag, Format::Png, source, &store) .unwrap(); - - assert_eq!(results[&Format::Html], "content→html"); + assert_eq!(executions.load(Ordering::SeqCst), 1); } #[test] - fn test_cached_node_is_skipped_on_second_run() { - let dir = tempfile::tempdir().unwrap(); - let cache_file = dir.path().join("dag-cache.json"); - - let dag = build_graph() - .build_multi_target_dag(Format::Markdown, &[Format::Html]) + fn failed_collection_transform_does_not_publish_partial_artifact() { + let dag = one_edge(Format::Png, Format::Pdf, InputKind::Collection); + let directory = tempfile::tempdir().unwrap(); + let store = ArtifactStore::new(directory.path().join("store")).unwrap(); + let source = store + .put_bytes( + b"page", + ArtifactDescriptor::for_format(Format::Png, ArtifactStorageClass::Source), + ) .unwrap(); - - // First run: populate the cache. - { - let mut executor = DagExecutor::new().with_cache(&cache_file); - executor.register_single( - Format::Markdown, - Format::Html, - arc_single(AppendTransform("→html".to_string())), - ); - executor - .execute(&dag, Format::Markdown, "content".to_string()) - .unwrap(); - } - - assert!( - cache_file.exists(), - "cache file must be written after first run" + let mut executor = DagExecutor::new(); + executor.register_aggregation( + Format::Png, + Format::Pdf, + Arc::new(FailingAggregation), ); - // Second run: transform should be served from cache. - let counter = Arc::new(CountingTransform::new()); - let mut executor = DagExecutor::new().with_cache(&cache_file); - executor.register_single( - Format::Markdown, - Format::Html, - Arc::clone(&counter) as Arc, + let before = count_artifact_objects(&store); + let result = executor.execute_artifacts( + &dag, + Format::Png, + ArtifactCollection::one(source), + &store, ); - - let results = executor - .execute(&dag, Format::Markdown, "content".to_string()) - .unwrap(); - - assert_eq!( - results[&Format::Html], - "content→html", - "output must match cached value" - ); - assert_eq!( - counter.call_count(), - 0, - "transform must not be called on cache hit" - ); - } - - #[test] - fn test_different_input_causes_cache_miss() { - let dir = tempfile::tempdir().unwrap(); - let cache_file = dir.path().join("dag-cache.json"); - - let dag = build_graph() - .build_multi_target_dag(Format::Markdown, &[Format::Html]) - .unwrap(); - - // Populate cache with "content A". - { - let mut executor = DagExecutor::new().with_cache(&cache_file); - executor.register_single( - Format::Markdown, - Format::Html, - arc_single(AppendTransform("→html".to_string())), - ); - executor - .execute(&dag, Format::Markdown, "content A".to_string()) - .unwrap(); + assert!(result.is_err()); + assert_eq!(before, count_artifact_objects(&store)); + } + + fn count_artifact_objects(store: &ArtifactStore) -> usize { + fn count_files(path: &Path) -> usize { + std::fs::read_dir(path) + .map(|entries| { + entries + .filter_map(std::result::Result::ok) + .map(|entry| { + let path = entry.path(); + if path.is_dir() { + count_files(&path) + } else { + usize::from(path.is_file()) + } + }) + .sum() + }) + .unwrap_or(0) } - // Run with different input "content B" — must not hit the cache. - let counter = Arc::new(CountingTransform::new()); - let mut executor = DagExecutor::new().with_cache(&cache_file); - executor.register_single( - Format::Markdown, - Format::Html, - Arc::clone(&counter) as Arc, - ); - - let results = executor - .execute(&dag, Format::Markdown, "content B".to_string()) - .unwrap(); - - assert_eq!(results[&Format::Html], "content B→counted"); - assert_eq!(counter.call_count(), 1, "transform must run on cache miss"); - } - - #[test] - fn test_no_cache_path_executes_normally() { - let dag = build_graph() - .build_multi_target_dag(Format::Markdown, &[Format::Html]) - .unwrap(); - - // Executor without a cache path — existing behaviour must be unchanged. - let mut executor = DagExecutor::new(); // no with_cache() - executor.register_single( - Format::Markdown, - Format::Html, - arc_single(AppendTransform("→html".to_string())), - ); - - let results = executor - .execute(&dag, Format::Markdown, "content".to_string()) - .unwrap(); - - assert_eq!(results[&Format::Html], "content→html"); + count_files(&store.root().join("objects/sha256")) } } diff --git a/crates/renderflow-core/src/lib.rs b/crates/renderflow-core/src/lib.rs index ea7905c..7389499 100644 --- a/crates/renderflow-core/src/lib.rs +++ b/crates/renderflow-core/src/lib.rs @@ -6,6 +6,7 @@ mod adapters; pub mod ai; pub mod app; +pub mod artifact; mod assets; mod audio; pub mod cache; diff --git a/docs/artifact-kernel.md b/docs/artifact-kernel.md new file mode 100644 index 0000000..444b96f --- /dev/null +++ b/docs/artifact-kernel.md @@ -0,0 +1,76 @@ +# Artifact kernel + +Renderflow's graph is format-oriented, but the original execution substrate moved every edge through UTF-8 `String` values. That was sufficient for early document transforms and unsafe as a universal media abstraction. The artifact kernel makes payloads file-backed and binary-safe while preserving the existing text-transform API during migration. + +## Model + +An `Artifact` records: + +- a stable artifact-record ID; +- canonical format and media type; +- SHA-256 payload digest and byte size; +- a store-relative payload handle; +- structured metadata; +- ordered source/parent artifact IDs; +- a lifecycle/storage class (`source`, `intermediate`, `terminal`, `cached`, or `ephemeral`). + +Payload identity and artifact-record identity are intentionally distinct. Payloads live at deterministic content-addressed paths derived only from their SHA-256 bytes. Artifact-record IDs also commit to canonical format and ordered source lineage. This allows two records to reuse the same stored bytes without collapsing distinct derivations. + +`ArtifactCollection` is an ordered first-class value. Aggregation transforms receive payload paths in collection order, which is required for operations such as pages → book, tracks → album, or frames → media package. + +## Store lifecycle + +`ArtifactStore` streams inputs into a temporary file while hashing them. A complete payload is atomically promoted to: + +```text +objects/sha256// +``` + +Canonical source files are imported rather than modified in place. Intermediate artifacts stay in the work store and do not need to appear beside final user outputs. Final materialization copies to a temporary file in the destination directory, syncs it, and only then atomically replaces the destination. + +A transform that fails after creating a temporary output cannot publish that partial file as a complete content-addressed artifact because import happens only after transform success. + +## Execution APIs + +`DagExecutor` now has an artifact-native substrate: + +- `execute_artifact(...)` for one source artifact; +- `execute_artifacts(...)` for an ordered source collection; +- `register_artifact(...)` for binary-safe transforms. + +The existing `execute(..., String)` and `register_single(..., Transform)` APIs remain compatibility surfaces. Existing transforms are wrapped by `TextTransformAdapter`; the adapter reads UTF-8 at the compatibility boundary, runs the legacy transform, then writes the result back into the artifact store with source lineage. + +Binary input sent to a legacy text transform fails explicitly instead of being lossily decoded. New binary transforms should implement `ArtifactTransform` until the versioned Transform v2/plugin contract is stabilized by issue #357. + +## Cache identity + +The artifact DAG cache no longer needs the complete in-memory input string to compute a node key. Its key commits to: + +- artifact-record ID; +- SHA-256 payload digest; +- byte size; +- source and target formats; +- transform cache identity. + +`register_single_with_identity(...)` and `ArtifactTransform::cache_identity()` provide a seam for configuration-aware identity. Full provider/tool/environment fingerprinting belongs to issues #357 and #359 and is not guessed by this kernel. + +Legacy string-cache files are treated as cache misses and replaced by the artifact cache schema after a successful run. + +## CLI graph builds + +Graph builds import the configured source into a hidden Renderflow state store before execution. The source is read as bytes, not UTF-8 text. Produced artifacts remain in the store until selected outputs are atomically materialized into the configured output directory. + +This removes the source and final-write UTF-8 assumptions from graph execution. A graph still needs an artifact-native transform for any binary edge; the kernel does not claim that every currently declared format has a production transform provider. + +## Flow boundary + +The kernel deliberately does not depend on `egohygiene/flow`. Its types contain the information needed to project a Renderflow artifact into Flow's artifact interchange contract later: stable ID, media type, SHA-256 digest, byte size, producer/metadata extension points, and ordered sources. Provenance-complete execution results and the concrete Flow projection are tracked separately by issues #355 and #358. + +## Migration sequence + +1. **#352 — artifact kernel:** binary-safe payloads, content-addressed storage, collections, atomic materialization, legacy text adapter. +2. **#357 — Transform v2/plugin SDK:** stabilize typed artifact I/O and execution context. +3. **#356/#359 — process/tool contracts:** central process policy and reproducible tool capability fingerprints. +4. **#355/#358 — evidence/resume:** provenance-complete results, checkpoints, and Flow provider seam. + +This sequence keeps existing document transforms working while moving the canonical execution substrate away from `String`. diff --git a/docs/user-guide/supported-formats.md b/docs/user-guide/supported-formats.md index 9f83670..34c052b 100644 --- a/docs/user-guide/supported-formats.md +++ b/docs/user-guide/supported-formats.md @@ -1,8 +1,9 @@ # Supported Formats !!! info - This page is generated from `src/input_format.rs`, `src/graph/format.rs`, - `src/audio/format.rs`, and `src/image/format.rs` by + This page is generated from `crates/renderflow-core/src/input_format.rs`, + `crates/renderflow-core/src/graph/format.rs`, `crates/renderflow-core/src/audio/format.rs`, + and `crates/renderflow-core/src/image/format.rs` by `scripts/generate_supported_formats_doc.py`. Do not edit it by hand. Renderflow recognizes format identifiers in four places: @@ -44,7 +45,28 @@ These are the canonical node names used in transform YAML files and graph output | `jpeg` | | `png` | | `tiff` | +| `webp` | +| `gif` | +| `bmp` | +| `avif` | +| `svg` | | `cbz` | +| `mp4` | +| `mov` | +| `mkv` | +| `webm` | +| `avi` | +| `json` | +| `yaml` | +| `toml` | +| `csv` | +| `tsv` | +| `xml` | +| `zip` | +| `tar.gz` | +| `tar.xz` | +| `srt` | +| `vtt` | ## Audio format identifiers diff --git a/scripts/generate_supported_formats_doc.py b/scripts/generate_supported_formats_doc.py old mode 100644 new mode 100755 index ab8928b..95efb70 --- a/scripts/generate_supported_formats_doc.py +++ b/scripts/generate_supported_formats_doc.py @@ -7,6 +7,7 @@ ROOT = Path(__file__).resolve().parents[1] +SOURCE_ROOT = ROOT / "crates" / "renderflow-core" / "src" DOC_PATH = ROOT / "docs" / "user-guide" / "supported-formats.md" @@ -105,10 +106,10 @@ def render_table(headers: list[str], rows: list[list[str]]) -> list[str]: def main() -> None: - input_text = read(ROOT / "src" / "input_format.rs") - graph_text = read(ROOT / "src" / "graph" / "format.rs") - audio_text = read(ROOT / "src" / "audio" / "format.rs") - image_text = read(ROOT / "src" / "image" / "format.rs") + input_text = read(SOURCE_ROOT / "input_format.rs") + graph_text = read(SOURCE_ROOT / "graph" / "format.rs") + audio_text = read(SOURCE_ROOT / "audio" / "format.rs") + image_text = read(SOURCE_ROOT / "image" / "format.rs") input_extensions = extract_input_extensions(input_text) @@ -180,8 +181,9 @@ def main() -> None: "# Supported Formats", "", "!!! info", - " This page is generated from `src/input_format.rs`, `src/graph/format.rs`,", - " `src/audio/format.rs`, and `src/image/format.rs` by", + " This page is generated from `crates/renderflow-core/src/input_format.rs`,", + " `crates/renderflow-core/src/graph/format.rs`, `crates/renderflow-core/src/audio/format.rs`,", + " and `crates/renderflow-core/src/image/format.rs` by", " `scripts/generate_supported_formats_doc.py`. Do not edit it by hand.", "", "Renderflow recognizes format identifiers in four places:",