From 84e5007f994560a64451cf32b79eaa6e25fee8f2 Mon Sep 17 00:00:00 2001 From: Alan Szmyt Date: Thu, 10 Sep 2026 12:15:18 -0400 Subject: [PATCH] feat: add provenance-complete execution evidence --- crates/renderflow-core/src/commands/build.rs | 20 +- crates/renderflow-core/src/evidence.rs | 598 ++++++++++++++++++ .../renderflow-core/src/graph/dag_executor.rs | 480 +++++++++++++- crates/renderflow-core/src/graph/mod.rs | 4 +- crates/renderflow-core/src/lib.rs | 8 +- crates/renderflow-core/src/planning.rs | 592 ++++++++++++++++- crates/renderflow-core/src/sdk.rs | 135 +++- .../execution-evidence/flow-artifact-v1.json | 19 + .../execution-evidence/outcome-matrix.json | 50 ++ docs/artifact-kernel.md | 2 +- docs/execution-evidence.md | 35 + mkdocs.yml | 1 + schemas/renderflow-run-v1.schema.json | 214 +++++++ 13 files changed, 2074 insertions(+), 84 deletions(-) create mode 100644 crates/renderflow-core/src/evidence.rs create mode 100644 crates/renderflow-core/tests/fixtures/execution-evidence/flow-artifact-v1.json create mode 100644 crates/renderflow-core/tests/fixtures/execution-evidence/outcome-matrix.json create mode 100644 docs/execution-evidence.md create mode 100644 schemas/renderflow-run-v1.schema.json diff --git a/crates/renderflow-core/src/commands/build.rs b/crates/renderflow-core/src/commands/build.rs index 6a0575b..b78053d 100644 --- a/crates/renderflow-core/src/commands/build.rs +++ b/crates/renderflow-core/src/commands/build.rs @@ -62,12 +62,30 @@ pub(crate) fn run_selection( // stdout is reserved for machine-readable plan evidence; tracing remains on stderr. println!("{}", serde_json::to_string_pretty(&result.plan)?); } - for output in &result.outputs { + for output in &result.run_manifest.artifact_manifest.outputs { if dry_run { info!("[DRY RUN] Planned output: {}", output); } else { info!("✔ Output written to: {}", output); } } + if let Some(manifest_path) = &result.manifest_path { + info!( + run_id = %result.run_manifest.run_id, + state = ?result.run_manifest.state, + "Run evidence written to: {}", + manifest_path + ); + } + if !result.is_success() { + anyhow::bail!( + "renderflow execution finished with state {:?}; inspect '{}' for structured evidence", + result.run_manifest.state, + result + .manifest_path + .as_deref() + .unwrap_or("renderflow-run.json") + ); + } Ok(()) } diff --git a/crates/renderflow-core/src/evidence.rs b/crates/renderflow-core/src/evidence.rs new file mode 100644 index 0000000..c753042 --- /dev/null +++ b/crates/renderflow-core/src/evidence.rs @@ -0,0 +1,598 @@ +//! Versioned, machine-readable execution evidence. +//! +//! Renderflow owns this richer native model. Integrations may project artifact +//! records into a supported interchange contract without making that external +//! contract part of the artifact kernel's internal representation. + +use std::collections::{BTreeMap, BTreeSet}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{SystemTime, UNIX_EPOCH}; + +use anyhow::Context; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; + +use crate::artifact::{Artifact, ArtifactStorageClass}; +use crate::toolchain::ToolchainSnapshot; + +pub const RUN_MANIFEST_SCHEMA_V1: &str = "renderflow.run/v1"; +pub const ARTIFACT_MANIFEST_SCHEMA_V1: &str = "renderflow.artifact-manifest/v1"; +pub const FLOW_ARTIFACT_SCHEMA_V1: &str = "flow.artifact/v1"; + +static RUN_SEQUENCE: AtomicU64 = AtomicU64::new(0); + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum RunState { + Planned, + Complete, + Partial, + Failed, + Cancelled, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum StepState { + Complete, + Reused, + Skipped, + Failed, + Cancelled, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum ValidationState { + Valid, + ValidWithWarnings, + Invalid, + Unavailable, + Skipped, + NotRequested, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum CacheDisposition { + Source, + Miss, + Hit, + NotApplicable, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum ArtifactRole { + Source, + Intermediate, + Terminal, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum FidelityDeclaration { + Lossless, + Partial, + Lossy, + PathDependent, + Unknown, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum DiagnosticSeverity { + Info, + Warning, + RecoverableFailure, + FatalFailure, + Cancelled, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct DigestEvidence { + pub algorithm: String, + pub value: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct ProducerEvidence { + pub system: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub transform: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub capability: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub provider: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub version: Option, +} + +impl ProducerEvidence { + pub fn source() -> Self { + Self { + system: "renderflow".to_string(), + transform: None, + capability: Some("artifact.source".to_string()), + provider: None, + version: Some(env!("CARGO_PKG_VERSION").to_string()), + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct ArtifactEvidence { + pub artifact_id: String, + /// Logical role from target/source intent (for example, `web` or `manuscript`). + pub role: String, + /// Lifecycle position inside this execution. + pub lifecycle: ArtifactRole, + pub locator: String, + pub format: String, + pub media_type: String, + pub digest: DigestEvidence, + pub size_bytes: u64, + pub producer: ProducerEvidence, + pub sources: Vec, + pub cache: CacheDisposition, + pub validation: ValidationState, + pub fidelity: FidelityDeclaration, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub warnings: Vec, + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + pub metadata: BTreeMap, +} + +impl ArtifactEvidence { + pub fn from_artifact( + artifact: &Artifact, + role: impl Into, + lifecycle: ArtifactRole, + locator: impl Into, + producer: ProducerEvidence, + validation: ValidationState, + fidelity: FidelityDeclaration, + ) -> Self { + let cache = match artifact.storage_class() { + ArtifactStorageClass::Source => CacheDisposition::Source, + ArtifactStorageClass::Cached => CacheDisposition::Hit, + ArtifactStorageClass::Intermediate + | ArtifactStorageClass::Terminal + | ArtifactStorageClass::Ephemeral => CacheDisposition::Miss, + }; + Self { + artifact_id: artifact.id().to_string(), + role: role.into(), + lifecycle, + locator: locator.into(), + format: artifact.format().to_string(), + media_type: artifact.media_type().to_string(), + digest: DigestEvidence { + algorithm: artifact.digest().algorithm().to_string(), + value: artifact.digest().value().to_string(), + }, + size_bytes: artifact.size_bytes(), + producer, + sources: artifact.sources().iter().map(ToString::to_string).collect(), + cache, + validation, + fidelity, + warnings: Vec::new(), + metadata: artifact + .metadata() + .iter() + .filter(|(key, _)| key.starts_with("renderflow.") && !is_sensitive_key(key)) + .map(|(key, value)| (key.clone(), value.clone())) + .collect(), + } + } + + pub fn to_flow_v1(&self) -> FlowArtifactV1 { + let mut seen_sources = BTreeSet::new(); + FlowArtifactV1 { + schema_version: FLOW_ARTIFACT_SCHEMA_V1.to_string(), + artifact_id: flow_artifact_id(&self.artifact_id), + role: self.role.clone(), + media_type: self.media_type.clone(), + digest: self.digest.clone(), + size_bytes: self.size_bytes, + producer: FlowProducerV1 { + owner: self.producer.system.clone(), + capability_id: self + .producer + .capability + .clone() + .or_else(|| self.producer.transform.clone()) + .unwrap_or_else(|| "artifact.produce".to_string()), + provider_version: self + .producer + .version + .clone() + .unwrap_or_else(|| "unknown".to_string()), + }, + sources: self + .sources + .iter() + .map(|source| flow_artifact_id(source)) + .filter(|source| seen_sources.insert(source.clone())) + .collect(), + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct StepEvidence { + pub step_id: String, + pub transform: String, + pub transform_version: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub capability: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub provider: Option, + pub input_artifacts: Vec, + pub output_artifacts: Vec, + pub configuration_digest: DigestEvidence, + pub started_at_unix_ms: u64, + pub completed_at_unix_ms: u64, + pub duration_ms: u64, + pub state: StepState, + pub cache: CacheDisposition, + pub validation: ValidationState, + pub fidelity: FidelityDeclaration, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub skip_reason: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub diagnostics: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct ExecutionDiagnostic { + pub severity: DiagnosticSeverity, + pub code: String, + pub message: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub step_id: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct ArtifactManifest { + pub schema_version: String, + pub run_id: String, + pub output_dir: String, + pub outputs: Vec, + pub artifacts: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct RunManifest { + pub schema_version: String, + pub run_id: String, + pub execution_plan_digest: DigestEvidence, + pub source_spec_digest: DigestEvidence, + pub engine_version: String, + pub started_at_unix_ms: u64, + pub completed_at_unix_ms: u64, + pub state: RunState, + pub artifact_manifest: ArtifactManifest, + pub steps: Vec, + #[serde(default)] + pub diagnostics: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub toolchain: Option, +} + +impl RunManifest { + pub fn flow_artifacts_v1(&self) -> Vec { + self.artifact_manifest + .artifacts + .iter() + .map(ArtifactEvidence::to_flow_v1) + .collect() + } + + pub fn cache_hits(&self) -> Vec { + self.steps + .iter() + .filter(|step| step.cache == CacheDisposition::Hit) + .flat_map(|step| step.output_artifacts.iter().cloned()) + .collect() + } + + pub fn skipped_transforms(&self) -> Vec { + self.steps + .iter() + .filter(|step| step.state == StepState::Skipped) + .map(|step| step.step_id.clone()) + .collect() + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct FlowProducerV1 { + pub owner: String, + pub capability_id: String, + pub provider_version: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct FlowArtifactV1 { + pub schema_version: String, + pub artifact_id: String, + pub role: String, + pub media_type: String, + pub digest: DigestEvidence, + pub size_bytes: u64, + pub producer: FlowProducerV1, + pub sources: Vec, +} + +pub fn unix_time_ms() -> u64 { + let millis = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis(); + u64::try_from(millis).unwrap_or(u64::MAX) +} + +pub fn sha256_serialized(value: &T) -> anyhow::Result { + let bytes = serde_json::to_vec(value).context("failed to serialize evidence digest input")?; + Ok(sha256_bytes(&bytes)) +} + +pub fn sha256_text(value: &str) -> DigestEvidence { + sha256_bytes(value.as_bytes()) +} + +/// Redact common credential assignments before provider errors enter durable evidence. +pub(crate) fn redact_sensitive_text(value: &str) -> String { + let mut redact_next = false; + value + .split_whitespace() + .map(|word| { + if redact_next { + if word.eq_ignore_ascii_case("bearer") { + return word.to_string(); + } + redact_next = false; + return "[REDACTED]".to_string(); + } + let lower = word.to_ascii_lowercase(); + if lower == "bearer" || lower.ends_with("authorization:") { + redact_next = true; + return word.to_string(); + } + for marker in [ + "api_key=", + "apikey=", + "token=", + "secret=", + "password=", + "credential=", + "authorization=", + ] { + if let Some(index) = lower.find(marker) { + return format!( + "{}{}[REDACTED]", + &word[..index], + &word[index..index + marker.len()] + ); + } + } + word.to_string() + }) + .collect::>() + .join(" ") +} + +fn is_sensitive_key(key: &str) -> bool { + let key = key.to_ascii_lowercase(); + [ + "api_key", + "apikey", + "token", + "secret", + "password", + "credential", + "authorization", + ] + .iter() + .any(|marker| key.contains(marker)) +} + +pub fn run_id(plan_digest: &DigestEvidence, started_at_unix_ms: u64) -> String { + let sequence = RUN_SEQUENCE.fetch_add(1, Ordering::Relaxed); + let material = format!( + "{}\0{}\0{}\0{}\0{}", + plan_digest.algorithm, + plan_digest.value, + started_at_unix_ms, + std::process::id(), + sequence + ); + let digest = sha256_text(&material); + format!("run:sha256:{}", digest.value) +} + +fn sha256_bytes(bytes: &[u8]) -> DigestEvidence { + let mut hasher = Sha256::new(); + hasher.update(bytes); + DigestEvidence { + algorithm: "sha256".to_string(), + value: format!("{:x}", hasher.finalize()), + } +} + +fn flow_artifact_id(native_id: &str) -> String { + let suffix = native_id + .strip_prefix("artifact:") + .unwrap_or(native_id) + .replace(':', "-") + .to_ascii_lowercase(); + let sanitized: String = suffix + .chars() + .map(|character| { + if character.is_ascii_lowercase() + || character.is_ascii_digit() + || matches!(character, '.' | '_' | '-') + { + character + } else { + '-' + } + }) + .collect(); + format!("artifact:{sanitized}") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[derive(Debug, Deserialize)] + struct OutcomeFixture { + name: String, + run_state: RunState, + step_state: StepState, + cache: CacheDisposition, + validation: ValidationState, + severity: DiagnosticSeverity, + } + + #[test] + fn flow_projection_uses_contract_safe_artifact_ids() { + assert_eq!( + flow_artifact_id("artifact:sha256:ABC123"), + "artifact:sha256-abc123" + ); + } + + #[test] + fn run_states_distinguish_non_successful_outcomes() { + let values = [ + RunState::Complete, + RunState::Partial, + RunState::Failed, + RunState::Cancelled, + ]; + let serialized = serde_json::to_string(&values).unwrap(); + assert!(serialized.contains("partial")); + assert!(serialized.contains("failed")); + assert!(serialized.contains("cancelled")); + } + + #[test] + fn flow_projection_matches_the_supported_producer_shape() { + let producer = FlowProducerV1 { + owner: "renderflow".to_string(), + capability_id: "document.render".to_string(), + provider_version: "0.2.1".to_string(), + }; + let value = serde_json::to_value(producer).unwrap(); + assert_eq!(value["owner"], "renderflow"); + assert_eq!(value["capability_id"], "document.render"); + assert_eq!(value["provider_version"], "0.2.1"); + assert!(value.get("system").is_none()); + assert!(value.get("capability").is_none()); + assert!(value.get("version").is_none()); + } + + #[test] + fn flow_projection_matches_pinned_v1_compatibility_fixture() { + let native = ArtifactEvidence { + artifact_id: "artifact:sha256:ABC123".to_string(), + role: "web".to_string(), + lifecycle: ArtifactRole::Terminal, + locator: "bundle:index.html".to_string(), + format: "html".to_string(), + media_type: "text/html".to_string(), + digest: DigestEvidence { + algorithm: "sha256".to_string(), + value: "0".repeat(64), + }, + size_bytes: 42, + producer: ProducerEvidence { + system: "renderflow".to_string(), + transform: Some("html-render".to_string()), + capability: Some("document.render".to_string()), + provider: Some("pandoc".to_string()), + version: Some("1.2.3".to_string()), + }, + sources: vec!["artifact:sha256:SOURCE123".to_string()], + cache: CacheDisposition::Miss, + validation: ValidationState::Valid, + fidelity: FidelityDeclaration::Lossless, + warnings: Vec::new(), + metadata: BTreeMap::new(), + }; + let expected: serde_json::Value = serde_json::from_str(include_str!( + "../tests/fixtures/execution-evidence/flow-artifact-v1.json" + )) + .unwrap(); + + assert_eq!(serde_json::to_value(native.to_flow_v1()).unwrap(), expected); + } + + #[test] + fn outcome_fixture_matrix_covers_required_states() { + let fixtures: Vec = serde_json::from_str(include_str!( + "../tests/fixtures/execution-evidence/outcome-matrix.json" + )) + .unwrap(); + assert_eq!(fixtures.len(), 6); + assert!(fixtures + .iter() + .any(|fixture| fixture.run_state == RunState::Partial)); + assert!(fixtures + .iter() + .any(|fixture| fixture.run_state == RunState::Cancelled)); + assert!(fixtures + .iter() + .any(|fixture| fixture.step_state == StepState::Reused)); + assert!(fixtures + .iter() + .any(|fixture| fixture.cache == CacheDisposition::Hit)); + assert!(fixtures + .iter() + .any(|fixture| fixture.validation == ValidationState::Invalid)); + assert!(fixtures + .iter() + .any(|fixture| fixture.severity == DiagnosticSeverity::FatalFailure)); + assert!(fixtures.iter().all(|fixture| !fixture.name.is_empty())); + } + + #[test] + fn durable_diagnostics_redact_common_secret_assignments() { + let redacted = redact_sensitive_text( + "provider failed api_key=super-secret Authorization: Bearer abc123 token=xyz", + ); + assert!(!redacted.contains("super-secret")); + assert!(!redacted.contains("abc123")); + assert!(!redacted.contains("xyz")); + assert!(redacted.contains("[REDACTED]")); + } + + #[test] + fn checked_in_run_schema_tracks_the_runtime_version() { + let schema: serde_json::Value = serde_json::from_str(include_str!( + "../../../schemas/renderflow-run-v1.schema.json" + )) + .unwrap(); + assert_eq!( + schema["properties"]["schema_version"]["const"], + RUN_MANIFEST_SCHEMA_V1 + ); + assert_eq!( + schema["$defs"]["artifact_manifest"]["properties"]["schema_version"]["const"], + ARTIFACT_MANIFEST_SCHEMA_V1 + ); + } +} diff --git a/crates/renderflow-core/src/graph/dag_executor.rs b/crates/renderflow-core/src/graph/dag_executor.rs index 7d2927a..cf0bd83 100644 --- a/crates/renderflow-core/src/graph/dag_executor.rs +++ b/crates/renderflow-core/src/graph/dag_executor.rs @@ -1,6 +1,7 @@ use std::collections::HashMap; use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; +use std::time::Instant; use anyhow::{Context, Result}; use rayon::prelude::*; @@ -12,6 +13,10 @@ use crate::artifact::{ ArtifactCollection, ArtifactDescriptor, ArtifactStorageClass, ArtifactStore, ArtifactTransform, TextTransformAdapter, }; +use crate::evidence::{ + redact_sensitive_text, sha256_text, unix_time_ms, CacheDisposition, DiagnosticSeverity, + ExecutionDiagnostic, FidelityDeclaration, StepEvidence, StepState, ValidationState, +}; use crate::transforms::aggregation::AggregationTransform; use crate::transforms::Transform; @@ -33,6 +38,169 @@ pub struct DagExecutor { max_parallel: Option, } +/// Artifact outputs and step evidence from one DAG execution. +pub struct DagExecutionReport { + pub artifacts: HashMap, + pub steps: Vec, + pub diagnostics: Vec, +} + +impl DagExecutionReport { + fn into_result(self) -> Result> { + ensure_steps_succeeded(&self.steps)?; + Ok(self.artifacts) + } +} + +struct CollectionExecutionReport { + artifacts: HashMap, + steps: Vec, + diagnostics: Vec, +} + +impl CollectionExecutionReport { + fn into_result(self) -> Result> { + ensure_steps_succeeded(&self.steps)?; + Ok(self.artifacts) + } +} + +struct ExecutedEdge { + format: Format, + artifacts: ArtifactCollection, + evidence: StepEvidence, +} + +struct SingleEdgeOutcome { + artifact: Artifact, + transform: String, + configuration_digest: crate::evidence::DigestEvidence, + cache: CacheDisposition, +} + +fn ensure_steps_succeeded(steps: &[StepEvidence]) -> Result<()> { + let has_failure = steps + .iter() + .any(|step| matches!(step.state, StepState::Failed | StepState::Cancelled)); + if !has_failure { + return Ok(()); + } + let failures = steps + .iter() + .filter(|step| matches!(step.state, StepState::Failed | StepState::Cancelled)) + .flat_map(|step| { + step.diagnostics + .iter() + .map(|diagnostic| diagnostic.message.clone()) + }) + .collect::>(); + if failures.is_empty() { + anyhow::bail!("artifact DAG execution failed without diagnostic detail") + } else { + anyhow::bail!("artifact DAG execution failed: {}", failures.join("; ")) + } +} + +fn edge_identity(edge: &TransformEdge) -> String { + edge.evidence + .get("transform_id") + .cloned() + .unwrap_or_else(|| format!("{}-to-{}", edge.from, edge.to)) +} + +fn edge_configuration_digest(edge: &TransformEdge) -> crate::evidence::DigestEvidence { + sha256_text(&format!( + "{}\0{}\0{}\0{}\0{:?}", + edge.from, + edge.to, + edge.provider_id.as_deref().unwrap_or(""), + edge.variant_id.as_deref().unwrap_or(""), + edge.evidence + )) +} + +fn edge_fidelity(edge: &TransformEdge) -> FidelityDeclaration { + if edge.input_kind.is_collection() { + FidelityDeclaration::PathDependent + } else if (edge.quality - 1.0).abs() < f32::EPSILON { + FidelityDeclaration::Lossless + } else { + FidelityDeclaration::Lossy + } +} + +fn failed_step( + edge: &TransformEdge, + inputs: Option<&ArtifactCollection>, + error: &anyhow::Error, + started_at_unix_ms: u64, + duration_ms: u64, +) -> StepEvidence { + let message = redact_sensitive_text(&error.to_string()); + let step_id = format!("step:{}-to-{}", edge.from, edge.to); + StepEvidence { + step_id: step_id.clone(), + transform: edge_identity(edge), + transform_version: edge + .variant_id + .clone() + .unwrap_or_else(|| "unknown".to_string()), + capability: edge.capability_id.clone(), + provider: edge.provider_id.clone(), + input_artifacts: inputs + .into_iter() + .flat_map(|collection| collection.iter()) + .map(|artifact| artifact.id().to_string()) + .collect(), + output_artifacts: Vec::new(), + configuration_digest: edge_configuration_digest(edge), + started_at_unix_ms, + completed_at_unix_ms: unix_time_ms(), + duration_ms, + state: StepState::Failed, + cache: CacheDisposition::Miss, + validation: if message.contains("returned artifact format") { + ValidationState::Invalid + } else { + ValidationState::Unavailable + }, + fidelity: edge_fidelity(edge), + skip_reason: None, + diagnostics: vec![ExecutionDiagnostic { + severity: DiagnosticSeverity::FatalFailure, + code: "execution.transform_failed".to_string(), + message, + step_id: Some(step_id), + }], + } +} + +fn skipped_step(edge: &TransformEdge, reason: &str) -> StepEvidence { + let timestamp = unix_time_ms(); + StepEvidence { + step_id: format!("step:{}-to-{}", edge.from, edge.to), + transform: edge_identity(edge), + transform_version: edge + .variant_id + .clone() + .unwrap_or_else(|| "unknown".to_string()), + capability: edge.capability_id.clone(), + provider: edge.provider_id.clone(), + input_artifacts: Vec::new(), + output_artifacts: Vec::new(), + configuration_digest: edge_configuration_digest(edge), + started_at_unix_ms: timestamp, + completed_at_unix_ms: timestamp, + duration_ms: 0, + state: StepState::Skipped, + cache: CacheDisposition::NotApplicable, + validation: ValidationState::Skipped, + fidelity: edge_fidelity(edge), + skip_reason: Some(reason.to_string()), + diagnostics: Vec::new(), + } +} + impl DagExecutor { /// Create an empty executor with no transforms registered. pub fn new() -> Self { @@ -176,14 +344,27 @@ impl DagExecutor { initial_artifact: Artifact, store: &ArtifactStore, ) -> Result> { - let collections = self.execute_artifacts( + self.execute_artifact_with_evidence(dag, source_format, initial_artifact, store)? + .into_result() + } + + /// Execute one artifact while retaining machine-readable step evidence. + pub fn execute_artifact_with_evidence( + &self, + dag: &MultiTargetDag, + source_format: Format, + initial_artifact: Artifact, + store: &ArtifactStore, + ) -> Result { + let report = self.execute_artifacts_with_evidence( dag, source_format, ArtifactCollection::one(initial_artifact), store, )?; - collections + let artifacts = report + .artifacts .into_iter() .map(|(format, collection)| { let artifact = collection.into_one().with_context(|| { @@ -194,7 +375,12 @@ impl DagExecutor { })?; Ok((format, artifact)) }) - .collect() + .collect::>>()?; + Ok(DagExecutionReport { + artifacts, + steps: report.steps, + diagnostics: report.diagnostics, + }) } /// Execute a DAG from an ordered source artifact collection. @@ -209,6 +395,17 @@ impl DagExecutor { initial_artifacts: ArtifactCollection, store: &ArtifactStore, ) -> Result> { + self.execute_artifacts_with_evidence(dag, source_format, initial_artifacts, store)? + .into_result() + } + + fn execute_artifacts_with_evidence( + &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"); } @@ -231,6 +428,8 @@ impl DagExecutor { let mut available: HashMap = HashMap::new(); available.insert(source_format, initial_artifacts); let mut remaining: Vec<&TransformEdge> = dag.execution_order(); + let mut steps = Vec::new(); + let mut diagnostics = Vec::new(); loop { let (wave, next_remaining): (Vec<_>, Vec<_>) = remaining @@ -243,6 +442,19 @@ impl DagExecutor { unreachable = next_remaining.len(), "Some DAG edges could not execute because their source format was never produced" ); + for edge in &next_remaining { + let step = skipped_step(edge, "required input artifact was not produced"); + diagnostics.push(ExecutionDiagnostic { + severity: DiagnosticSeverity::RecoverableFailure, + code: "execution.step_skipped".to_string(), + message: format!( + "Skipped transform {} → {} because its input was unavailable", + edge.from, edge.to + ), + step_id: Some(step.step_id.clone()), + }); + steps.push(step); + } } break; } @@ -250,8 +462,17 @@ impl DagExecutor { debug!(wave_size = wave.len(), "Executing artifact DAG wave"); let execute_wave = || { wave.into_par_iter() - .map(|edge| self.execute_edge(edge, &available, store, cache.as_ref())) - .collect::>>() + .map(|edge| { + let started_at_unix_ms = unix_time_ms(); + let started = Instant::now(); + ( + edge, + self.execute_edge(edge, &available, store, cache.as_ref()), + started_at_unix_ms, + u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX), + ) + }) + .collect::>() }; let wave_results = if let Some(pool) = &thread_pool { pool.install(execute_wave) @@ -259,8 +480,33 @@ impl DagExecutor { execute_wave() }; - for (format, artifacts) in wave_results? { - available.insert(format, artifacts); + for (edge, outcome, started_at_unix_ms, duration_ms) in wave_results { + match outcome { + Ok(executed) => { + available.insert(executed.format, executed.artifacts); + steps.push(executed.evidence); + } + Err(error) => { + let step = failed_step( + edge, + available.get(&edge.from), + &error, + started_at_unix_ms, + duration_ms, + ); + diagnostics.push(ExecutionDiagnostic { + severity: DiagnosticSeverity::FatalFailure, + code: "execution.transform_failed".to_string(), + message: step + .diagnostics + .first() + .map(|diagnostic| diagnostic.message.clone()) + .unwrap_or_else(|| "Transform failed".to_string()), + step_id: Some(step.step_id.clone()), + }); + steps.push(step); + } + } } remaining = next_remaining; } @@ -285,7 +531,11 @@ impl DagExecutor { } } - Ok(available) + Ok(CollectionExecutionReport { + artifacts: available, + steps, + diagnostics, + }) } fn execute_edge( @@ -294,7 +544,9 @@ impl DagExecutor { available: &HashMap, store: &ArtifactStore, cache: Option<&Mutex>, - ) -> Result<(Format, ArtifactCollection)> { + ) -> Result { + let started_at_unix_ms = unix_time_ms(); + let started = Instant::now(); let inputs = available.get(&edge.from).ok_or_else(|| { anyhow::anyhow!( "Source format '{}' was not available for DAG edge", @@ -302,19 +554,72 @@ impl DagExecutor { ) })?; - if edge.input_kind.is_single() { + let input_artifacts = inputs + .iter() + .map(|artifact| artifact.id().to_string()) + .collect::>(); + let (artifacts, transform, configuration_digest, cache) = if edge.input_kind.is_single() { 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))) + let outcome = self.execute_single_edge(edge, &input, store, cache)?; + ( + ArtifactCollection::one(outcome.artifact), + outcome.transform, + outcome.configuration_digest, + outcome.cache, + ) } else { - let output = self.execute_collection_edge(edge, inputs, store)?; - Ok((edge.to, ArtifactCollection::one(output))) - } + let (output, transform) = self.execute_collection_edge(edge, inputs, store)?; + ( + ArtifactCollection::one(output), + transform.clone(), + sha256_text(&transform), + CacheDisposition::NotApplicable, + ) + }; + let completed_at_unix_ms = unix_time_ms(); + let output_artifacts = artifacts + .iter() + .map(|artifact| artifact.id().to_string()) + .collect(); + let fidelity = if edge.input_kind.is_collection() { + FidelityDeclaration::PathDependent + } else if (edge.quality - 1.0).abs() < f32::EPSILON { + FidelityDeclaration::Lossless + } else { + FidelityDeclaration::Lossy + }; + Ok(ExecutedEdge { + format: edge.to, + artifacts, + evidence: StepEvidence { + step_id: format!("step:{}-to-{}", edge.from, edge.to), + transform, + transform_version: "unstable-v1".to_string(), + capability: edge.capability_id.clone(), + provider: edge.provider_id.clone(), + input_artifacts, + output_artifacts, + configuration_digest, + started_at_unix_ms, + completed_at_unix_ms, + duration_ms: u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX), + state: if cache == CacheDisposition::Hit { + StepState::Reused + } else { + StepState::Complete + }, + cache, + validation: ValidationState::Valid, + fidelity, + skip_reason: None, + diagnostics: Vec::new(), + }, + }) } fn execute_single_edge( @@ -323,7 +628,7 @@ impl DagExecutor { input: &Artifact, store: &ArtifactStore, cache: Option<&Mutex>, - ) -> Result { + ) -> Result { let transform = self .single_transforms .get(&(edge.from, edge.to)) @@ -351,9 +656,14 @@ impl DagExecutor { artifact = %cached.id(), "Artifact cache hit; skipping transform" ); - return Ok(cached - .clone() - .with_storage_class(ArtifactStorageClass::Cached)); + return Ok(SingleEdgeOutcome { + artifact: cached + .clone() + .with_storage_class(ArtifactStorageClass::Cached), + transform: transform.name().to_string(), + configuration_digest: sha256_text(&cache_identity), + cache: CacheDisposition::Hit, + }); } } } @@ -380,7 +690,12 @@ impl DagExecutor { guard.insert(cache_key, output.clone()); } } - Ok(output) + Ok(SingleEdgeOutcome { + artifact: output, + transform: transform.name().to_string(), + configuration_digest: sha256_text(&cache_identity), + cache: CacheDisposition::Miss, + }) } fn execute_collection_edge( @@ -388,7 +703,7 @@ impl DagExecutor { edge: &TransformEdge, inputs: &ArtifactCollection, store: &ArtifactStore, - ) -> Result { + ) -> Result<(Artifact, String)> { let transform = self .aggregation_transforms .get(&(edge.from, edge.to)) @@ -453,7 +768,7 @@ impl DagExecutor { .with_metadata("renderflow.transform", transform.name()), )?; self.validate_output_format(edge, &output)?; - Ok(output) + Ok((output, transform.name().to_string())) } fn validate_output_format(&self, edge: &TransformEdge, output: &Artifact) -> Result<()> { @@ -589,6 +904,28 @@ mod tests { } } + struct WrongFormatTransform; + + impl ArtifactTransform for WrongFormatTransform { + fn name(&self) -> &str { + "wrong-format" + } + + 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(Format::Png, ArtifactStorageClass::Intermediate) + .with_source(input.id().clone()), + ) + } + } + fn one_edge(from: Format, to: Format, input_kind: InputKind) -> MultiTargetDag { let mut graph = TransformGraph::new(); graph.add_transform(TransformEdge::with_input_kind( @@ -702,6 +1039,105 @@ mod tests { assert_eq!(executions.load(Ordering::SeqCst), 1); } + #[test] + fn execution_evidence_reports_cache_miss_then_hit() { + 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), + ) + .unwrap(); + let mut executor = DagExecutor::new().with_cache(&cache_path); + executor.register_artifact( + Format::Png, + Format::Webp, + Arc::new(CountingBinaryTransform { + executions: Arc::new(AtomicUsize::new(0)), + }), + ); + + let first = executor + .execute_artifact_with_evidence(&dag, Format::Png, source.clone(), &store) + .unwrap(); + let second = executor + .execute_artifact_with_evidence(&dag, Format::Png, source, &store) + .unwrap(); + + assert_eq!(first.steps[0].cache, CacheDisposition::Miss); + assert_eq!(first.steps[0].state, StepState::Complete); + assert_eq!(second.steps[0].cache, CacheDisposition::Hit); + assert_eq!(second.steps[0].state, StepState::Reused); + assert!(second.diagnostics.is_empty()); + } + + #[test] + fn partial_execution_records_failure_and_downstream_skip() { + let mut graph = TransformGraph::new(); + graph.add_transform(TransformEdge::new(Format::Png, Format::Webp, 1.0, 1.0)); + graph.add_collection_transform(Format::Png, Format::Pdf, 1.0, 1.0); + graph.add_transform(TransformEdge::new(Format::Pdf, Format::Html, 1.0, 1.0)); + let dag = graph + .build_multi_target_dag(Format::Png, &[Format::Webp, Format::Html]) + .unwrap(); + 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(); + let mut executor = DagExecutor::new(); + executor.register_artifact(Format::Png, Format::Webp, Arc::new(BinaryCopyTransform)); + executor.register_aggregation(Format::Png, Format::Pdf, Arc::new(FailingAggregation)); + + let report = executor + .execute_artifact_with_evidence(&dag, Format::Png, source, &store) + .unwrap(); + + assert!(report.artifacts.contains_key(&Format::Webp)); + assert!(!report.artifacts.contains_key(&Format::Html)); + assert!(report + .steps + .iter() + .any(|step| step.state == StepState::Complete)); + assert!(report + .steps + .iter() + .any(|step| step.state == StepState::Failed)); + assert!(report + .steps + .iter() + .any(|step| step.state == StepState::Skipped)); + } + + #[test] + fn output_contract_validation_failure_is_structured() { + 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 source = store + .put_bytes( + b"image", + ArtifactDescriptor::for_format(Format::Png, ArtifactStorageClass::Source), + ) + .unwrap(); + let mut executor = DagExecutor::new(); + executor.register_artifact(Format::Png, Format::Webp, Arc::new(WrongFormatTransform)); + + let report = executor + .execute_artifact_with_evidence(&dag, Format::Png, source, &store) + .unwrap(); + + assert_eq!(report.steps[0].state, StepState::Failed); + assert_eq!(report.steps[0].validation, ValidationState::Invalid); + assert_eq!(report.diagnostics[0].code, "execution.transform_failed"); + } + #[test] fn failed_collection_transform_does_not_publish_partial_artifact() { let dag = one_edge(Format::Png, Format::Pdf, InputKind::Collection); diff --git a/crates/renderflow-core/src/graph/mod.rs b/crates/renderflow-core/src/graph/mod.rs index b1fa0bf..f98670e 100644 --- a/crates/renderflow-core/src/graph/mod.rs +++ b/crates/renderflow-core/src/graph/mod.rs @@ -10,10 +10,10 @@ mod pathfinding; pub mod renderers; mod transform_edge; -pub use dag_executor::DagExecutor; +pub use dag_executor::{DagExecutionReport, DagExecutor}; pub use definition::TransformDefinition; pub use definition_registry::TransformDefinitionRegistry; -pub use execution_plan::ExecutionPlan; +pub use execution_plan::{DiagnosticLevel, ExecutionPlan}; pub use format::Format; pub use input_kind::InputKind; pub use multi_target::MultiTargetDag; diff --git a/crates/renderflow-core/src/lib.rs b/crates/renderflow-core/src/lib.rs index 6df9676..ec639dc 100644 --- a/crates/renderflow-core/src/lib.rs +++ b/crates/renderflow-core/src/lib.rs @@ -15,6 +15,7 @@ mod commands; mod compat; mod config; pub mod detect; +pub mod evidence; pub mod error; pub mod graph; mod image; @@ -30,8 +31,9 @@ pub mod super_resolution; pub mod toolchain; pub mod transforms; +pub use evidence::{ArtifactManifest, RunManifest}; pub use sdk::{ - ArtifactManifest, ArtifactProfile, CancellationToken, DiagnosticReport, Engine, EngineBuilder, - ExecutionRequest, ExecutionResult, InspectionRequest, PlanRequest, ProgressEvent, - ProgressReporter, ProgressStage, RenderflowError, + ArtifactProfile, CancellationToken, DiagnosticReport, Engine, EngineBuilder, ExecutionRequest, + ExecutionResult, InspectionRequest, PlanRequest, ProgressEvent, ProgressReporter, + ProgressStage, RenderflowError, }; diff --git a/crates/renderflow-core/src/planning.rs b/crates/renderflow-core/src/planning.rs index c8f69c1..360c6d2 100644 --- a/crates/renderflow-core/src/planning.rs +++ b/crates/renderflow-core/src/planning.rs @@ -6,6 +6,7 @@ use std::collections::{BTreeSet, HashMap, HashSet}; use std::fs; +use std::io::Write; use std::path::{Component, Path, PathBuf}; use std::sync::Arc; @@ -14,10 +15,17 @@ use anyhow::{Context, Result}; use crate::adapters::strategy::{ document_input_format, output_type_for_format, StrategyArtifactTransform, }; -use crate::artifact::{ArtifactDescriptor, ArtifactStorageClass, ArtifactStore}; +use crate::artifact::{Artifact, ArtifactDescriptor, ArtifactStorageClass, ArtifactStore}; +use crate::evidence::{ + redact_sensitive_text, run_id, sha256_serialized, unix_time_ms, ArtifactEvidence, + ArtifactManifest, ArtifactRole, DiagnosticSeverity, ExecutionDiagnostic, FidelityDeclaration, + ProducerEvidence, RunManifest, RunState, StepEvidence, StepState, ValidationState, + ARTIFACT_MANIFEST_SCHEMA_V1, RUN_MANIFEST_SCHEMA_V1, +}; use crate::graph::capability::{FormatCapabilityRegistry, FormatFamily}; use crate::graph::{ - DagExecutor, ExecutionPlan, Format, MultiTargetDag, TransformEdge, TransformGraph, + DagExecutionReport, DagExecutor, DiagnosticLevel, ExecutionPlan, Format, MultiTargetDag, + TransformEdge, TransformGraph, }; use crate::optimization::OptimizationMode; use crate::spec::{ @@ -162,6 +170,19 @@ pub struct CanonicalExecutionResult { pub outputs: Vec, pub diagnostics: Vec, pub toolchain: Option, + /// Authoritative evidence derived from the actual executor outcome. + pub run_manifest: RunManifest, + /// Persisted run-manifest path. Dry runs are side-effect free and return `None`. + pub manifest_path: Option, +} + +impl CanonicalExecutionResult { + pub fn is_success(&self) -> bool { + matches!( + self.run_manifest.state, + RunState::Planned | RunState::Complete + ) + } } pub fn resolve(request: PlanningRequest) -> Result { @@ -314,15 +335,25 @@ pub fn resolve(request: PlanningRequest) -> Result { } pub fn execute(mut resolved: ResolvedExecution, dry_run: bool) -> Result { + let started_at_unix_ms = unix_time_ms(); let predicted = resolved.predicted_output_paths()?; if dry_run { - return Ok(CanonicalExecutionResult { - plan: resolved.plan.clone(), - output_dir: resolved.spec.output.bundle_root.clone(), - outputs: predicted + let run_manifest = build_run_manifest( + &resolved, + started_at_unix_ms, + RunState::Planned, + predicted .iter() .map(|path| path.display().to_string()) .collect(), + Vec::new(), + Vec::new(), + plan_diagnostics(&resolved), + )?; + return Ok(CanonicalExecutionResult { + plan: resolved.plan.clone(), + output_dir: resolved.spec.output.bundle_root.clone(), + outputs: run_manifest.artifact_manifest.outputs.clone(), diagnostics: resolved .plan .diagnostics @@ -330,12 +361,11 @@ pub fn execute(mut resolved: ResolvedExecution, dry_run: bool) -> Result Result store, + Err(error) => { + return failed_execution_result( + &resolved, + &output_root, + started_at_unix_ms, + Vec::new(), + Vec::new(), + "execution.artifact_store_failed", + error, + ) + } + }; + let source_artifact = match store.import_path( &resolved.source_path, ArtifactDescriptor::for_format(resolved.source_format, ArtifactStorageClass::Source) .with_metadata("renderflow.source_id", resolved.source.id.clone()), - )?; + ) { + Ok(artifact) => artifact, + Err(error) => { + return failed_execution_result( + &resolved, + &output_root, + started_at_unix_ms, + Vec::new(), + Vec::new(), + "execution.source_import_failed", + error, + ) + } + }; let executor = std::mem::take(&mut resolved.executor); let mut executor = executor @@ -363,52 +444,493 @@ pub fn execute(mut resolved: ResolvedExecution, dry_run: bool) -> Result report, + Err(error) => { + let source_evidence = source_artifact_evidence(&resolved, &source_artifact); + return failed_execution_result( + &resolved, + &output_root, + started_at_unix_ms, + vec![source_evidence], + Vec::new(), + "execution.executor_failed", + error, + ); + } + }; + enrich_step_versions(&mut report.steps, resolved.plan.toolchain.as_ref()); + + let mut diagnostics = plan_diagnostics(&resolved); + diagnostics.append(&mut report.diagnostics); + let mut output_locators = HashMap::::new(); + let mut actual_outputs = Vec::new(); + let mut target_failures = false; + + if let Err(error) = validate_post_execution_budgets(&resolved, &report.artifacts) { + target_failures = true; + diagnostics.push(ExecutionDiagnostic { + severity: DiagnosticSeverity::FatalFailure, + code: "execution.post_budget_failed".to_string(), + message: redact_sensitive_text(&error.to_string()), + step_id: None, + }); + } - validate_post_execution_budgets(&resolved, &artifacts)?; for (target, destination) in resolved.targets.iter().zip(predicted.iter()) { - let artifact = artifacts.get(&target.format).ok_or_else(|| { - anyhow::anyhow!( - "execution plan completed without producing selected target '{}'", - target.format - ) - })?; + let Some(artifact) = report.artifacts.get(&target.format) else { + target_failures = true; + diagnostics.push(ExecutionDiagnostic { + severity: DiagnosticSeverity::RecoverableFailure, + code: "execution.target_missing".to_string(), + message: format!( + "Execution did not produce selected target '{}'", + target.format + ), + step_id: None, + }); + continue; + }; if resolved.spec.execution.validation.required && artifact.size_bytes() == 0 { - anyhow::bail!( - "validation failed: target '{}' produced an empty artifact", - target.format - ); + target_failures = true; + diagnostics.push(ExecutionDiagnostic { + severity: DiagnosticSeverity::FatalFailure, + code: "validation.empty_artifact".to_string(), + message: format!("Target '{}' produced an empty artifact", target.format), + step_id: producing_step_id(artifact, &report.steps), + }); + continue; + } + if target_failures + && diagnostics + .iter() + .any(|diagnostic| diagnostic.code == "execution.post_budget_failed") + { + continue; } - store.materialize(artifact, destination)?; + if let Err(error) = store.materialize(artifact, destination) { + target_failures = true; + diagnostics.push(ExecutionDiagnostic { + severity: DiagnosticSeverity::RecoverableFailure, + code: "execution.materialization_failed".to_string(), + message: redact_sensitive_text(&error.to_string()), + step_id: producing_step_id(artifact, &report.steps), + }); + continue; + } + let locator = bundle_locator(&output_root, destination); + output_locators.insert(artifact.id().to_string(), locator); + actual_outputs.push(destination.display().to_string()); } + let artifacts = artifact_evidence( + &resolved, + &source_artifact, + &report, + &output_locators, + &diagnostics, + ); + let has_failed_step = report + .steps + .iter() + .any(|step| step.state == StepState::Failed); + let has_failure = target_failures || has_failed_step; + let state = if has_failure && actual_outputs.is_empty() { + RunState::Failed + } else if has_failure { + RunState::Partial + } else { + RunState::Complete + }; + let run_manifest = build_run_manifest( + &resolved, + started_at_unix_ms, + state, + actual_outputs.clone(), + artifacts, + report.steps, + diagnostics, + )?; + let manifest_path = persist_run_manifest(&output_root, &run_manifest)?; + Ok(CanonicalExecutionResult { plan: resolved.plan.clone(), output_dir: resolved.spec.output.bundle_root.clone(), - outputs: predicted + outputs: actual_outputs, + diagnostics: run_manifest + .diagnostics .iter() - .map(|path| path.display().to_string()) + .map(|diagnostic| diagnostic.message.clone()) .collect(), - diagnostics: resolved - .plan + toolchain: resolved.plan.toolchain.clone(), + run_manifest, + manifest_path: Some(manifest_path.display().to_string()), + }) +} + +/// Record a cancellation that occurs after planning but before transform execution. +pub fn cancelled(resolved: ResolvedExecution) -> Result { + let started_at_unix_ms = unix_time_ms(); + let output_root = PathBuf::from(&resolved.spec.output.bundle_root); + fs::create_dir_all(&output_root).with_context(|| { + format!( + "failed to create output directory '{}' for cancellation evidence", + output_root.display() + ) + })?; + let mut diagnostics = plan_diagnostics(&resolved); + diagnostics.push(ExecutionDiagnostic { + severity: DiagnosticSeverity::Cancelled, + code: "execution.cancelled".to_string(), + message: "Execution was cancelled before transforms started".to_string(), + step_id: None, + }); + let mut cancelled_steps = resolved + .plan + .edges + .iter() + .map(|edge| { + Ok(StepEvidence { + step_id: format!("step:{}-to-{}", edge.from, edge.to), + transform: edge + .evidence + .get("transform_id") + .cloned() + .unwrap_or_else(|| format!("{}-to-{}", edge.from, edge.to)), + transform_version: "unknown".to_string(), + capability: edge.capability_id.clone(), + provider: edge.provider_id.clone(), + input_artifacts: Vec::new(), + output_artifacts: Vec::new(), + configuration_digest: sha256_serialized(edge)?, + started_at_unix_ms, + completed_at_unix_ms: started_at_unix_ms, + duration_ms: 0, + state: StepState::Cancelled, + cache: crate::evidence::CacheDisposition::NotApplicable, + validation: ValidationState::Skipped, + fidelity: FidelityDeclaration::Unknown, + skip_reason: Some("execution cancelled before transform started".to_string()), + diagnostics: Vec::new(), + }) + }) + .collect::>>()?; + enrich_step_versions(&mut cancelled_steps, resolved.plan.toolchain.as_ref()); + let run_manifest = build_run_manifest( + &resolved, + started_at_unix_ms, + RunState::Cancelled, + Vec::new(), + Vec::new(), + cancelled_steps, + diagnostics, + )?; + let manifest_path = persist_run_manifest(&output_root, &run_manifest)?; + Ok(canonical_result( + &resolved, + run_manifest, + Some(manifest_path.display().to_string()), + Vec::new(), + )) +} + +fn failed_execution_result( + resolved: &ResolvedExecution, + output_root: &Path, + started_at_unix_ms: u64, + artifacts: Vec, + steps: Vec, + code: &str, + error: anyhow::Error, +) -> Result { + let mut diagnostics = plan_diagnostics(resolved); + diagnostics.push(ExecutionDiagnostic { + severity: DiagnosticSeverity::FatalFailure, + code: code.to_string(), + message: redact_sensitive_text(&error.to_string()), + step_id: None, + }); + let run_manifest = build_run_manifest( + resolved, + started_at_unix_ms, + RunState::Failed, + Vec::new(), + artifacts, + steps, + diagnostics, + )?; + let manifest_path = persist_run_manifest(output_root, &run_manifest)?; + Ok(canonical_result( + resolved, + run_manifest, + Some(manifest_path.display().to_string()), + Vec::new(), + )) +} + +fn canonical_result( + resolved: &ResolvedExecution, + run_manifest: RunManifest, + manifest_path: Option, + outputs: Vec, +) -> CanonicalExecutionResult { + CanonicalExecutionResult { + plan: resolved.plan.clone(), + output_dir: resolved.spec.output.bundle_root.clone(), + outputs, + diagnostics: run_manifest .diagnostics .iter() .map(|diagnostic| diagnostic.message.clone()) .collect(), toolchain: resolved.plan.toolchain.clone(), + run_manifest, + manifest_path, + } +} + +#[allow(clippy::too_many_arguments)] +fn build_run_manifest( + resolved: &ResolvedExecution, + started_at_unix_ms: u64, + state: RunState, + outputs: Vec, + artifacts: Vec, + steps: Vec, + diagnostics: Vec, +) -> Result { + let execution_plan_digest = sha256_serialized(&resolved.plan)?; + let source_spec_digest = sha256_serialized(&resolved.spec)?; + let run_id = run_id(&execution_plan_digest, started_at_unix_ms); + Ok(RunManifest { + schema_version: RUN_MANIFEST_SCHEMA_V1.to_string(), + run_id: run_id.clone(), + execution_plan_digest, + source_spec_digest, + engine_version: env!("CARGO_PKG_VERSION").to_string(), + started_at_unix_ms, + completed_at_unix_ms: unix_time_ms(), + state, + artifact_manifest: ArtifactManifest { + schema_version: ARTIFACT_MANIFEST_SCHEMA_V1.to_string(), + run_id, + output_dir: resolved.spec.output.bundle_root.clone(), + outputs, + artifacts, + }, + steps, + diagnostics, + toolchain: resolved.plan.toolchain.clone(), }) } +fn persist_run_manifest(output_root: &Path, manifest: &RunManifest) -> Result { + let destination = output_root.join("renderflow-run.json"); + let mut temporary = tempfile::NamedTempFile::new_in(output_root).with_context(|| { + format!( + "failed to create temporary run manifest in '{}'", + output_root.display() + ) + })?; + temporary + .write_all(&serde_json::to_vec_pretty(manifest)?) + .context("failed to write run manifest")?; + temporary.flush().context("failed to flush run manifest")?; + temporary + .as_file() + .sync_all() + .context("failed to sync run manifest")?; + temporary + .persist(&destination) + .map_err(|error| error.error) + .with_context(|| { + format!( + "failed to atomically persist run manifest '{}'", + destination.display() + ) + })?; + Ok(destination) +} + +fn plan_diagnostics(resolved: &ResolvedExecution) -> Vec { + resolved + .plan + .diagnostics + .iter() + .map(|diagnostic| ExecutionDiagnostic { + severity: match &diagnostic.level { + DiagnosticLevel::Info => DiagnosticSeverity::Info, + DiagnosticLevel::Warning => DiagnosticSeverity::Warning, + DiagnosticLevel::Error => DiagnosticSeverity::FatalFailure, + }, + code: "planning.diagnostic".to_string(), + message: diagnostic.message.clone(), + step_id: None, + }) + .collect() +} + +fn source_artifact_evidence(resolved: &ResolvedExecution, source: &Artifact) -> ArtifactEvidence { + ArtifactEvidence::from_artifact( + source, + resolved + .source + .role + .clone() + .unwrap_or_else(|| resolved.source.id.clone()), + ArtifactRole::Source, + artifact_store_locator(source), + ProducerEvidence::source(), + ValidationState::NotRequested, + FidelityDeclaration::Lossless, + ) +} + +fn artifact_evidence( + resolved: &ResolvedExecution, + source: &Artifact, + report: &DagExecutionReport, + output_locators: &HashMap, + diagnostics: &[ExecutionDiagnostic], +) -> Vec { + let mut evidence = vec![source_artifact_evidence(resolved, source)]; + let mut artifacts = report.artifacts.iter().collect::>(); + artifacts.sort_by(|(left_format, left), (right_format, right)| { + left_format + .to_string() + .cmp(&right_format.to_string()) + .then_with(|| left.id().as_str().cmp(right.id().as_str())) + }); + for (format, artifact) in artifacts { + if artifact.id() == source.id() { + continue; + } + let target = resolved + .targets + .iter() + .find(|target| target.format == *format); + let lifecycle = if target.is_some() { + ArtifactRole::Terminal + } else { + ArtifactRole::Intermediate + }; + let role = target + .and_then(|target| target.role.clone().or_else(|| target.id.clone())) + .unwrap_or_else(|| artifact.format().to_string()); + let producing_step = report.steps.iter().find(|step| { + step.output_artifacts + .iter() + .any(|artifact_id| artifact_id == artifact.id().as_str()) + }); + let invalid = producing_step.is_some_and(|step| { + diagnostics.iter().any(|diagnostic| { + diagnostic.code.starts_with("validation.") + && diagnostic.step_id.as_deref() == Some(step.step_id.as_str()) + }) + }); + let validation = if invalid { + ValidationState::Invalid + } else if resolved.spec.execution.validation.required && target.is_some() { + ValidationState::Valid + } else { + ValidationState::NotRequested + }; + let producer = producing_step + .map(|step| ProducerEvidence { + system: "renderflow".to_string(), + transform: Some(step.transform.clone()), + capability: step.capability.clone(), + provider: step.provider.clone(), + version: Some(step.transform_version.clone()), + }) + .unwrap_or_else(ProducerEvidence::source); + let fidelity = producing_step + .map(|step| step.fidelity) + .unwrap_or(FidelityDeclaration::Unknown); + let locator = output_locators + .get(artifact.id().as_str()) + .cloned() + .unwrap_or_else(|| artifact_store_locator(artifact)); + let mut artifact_evidence = ArtifactEvidence::from_artifact( + artifact, role, lifecycle, locator, producer, validation, fidelity, + ); + if let Some(step) = producing_step { + artifact_evidence.warnings = diagnostics + .iter() + .filter(|diagnostic| { + diagnostic.severity == DiagnosticSeverity::Warning + && diagnostic.step_id.as_deref() == Some(step.step_id.as_str()) + }) + .map(|diagnostic| diagnostic.message.clone()) + .collect(); + } + evidence.push(artifact_evidence); + } + evidence +} + +fn artifact_store_locator(artifact: &Artifact) -> String { + format!( + "artifact-store:{}", + artifact + .payload() + .relative_path() + .to_string_lossy() + .replace('\\', "/") + ) +} + +fn bundle_locator(output_root: &Path, destination: &Path) -> String { + let relative = destination.strip_prefix(output_root).unwrap_or(destination); + format!("bundle:{}", relative.to_string_lossy().replace('\\', "/")) +} + +fn producing_step_id(artifact: &Artifact, steps: &[StepEvidence]) -> Option { + steps + .iter() + .find(|step| { + step.output_artifacts + .iter() + .any(|artifact_id| artifact_id == artifact.id().as_str()) + }) + .map(|step| step.step_id.clone()) +} + +fn enrich_step_versions(steps: &mut [StepEvidence], toolchain: Option<&ToolchainSnapshot>) { + for step in steps { + let version = step.provider.as_deref().and_then(|provider| { + toolchain.and_then(|snapshot| { + snapshot + .selected_tools + .iter() + .find(|tool| tool.id.as_str() == provider) + .and_then(|tool| tool.version.clone()) + }) + }); + step.transform_version = version.unwrap_or_else(|| env!("CARGO_PKG_VERSION").to_string()); + } +} + fn apply_request_overrides(spec: &mut SpecV2, request: &PlanningRequest) -> Result<()> { if let Some(optimization) = request.optimization { spec.execution.optimization = optimization; diff --git a/crates/renderflow-core/src/sdk.rs b/crates/renderflow-core/src/sdk.rs index c3578ed..e6b3557 100644 --- a/crates/renderflow-core/src/sdk.rs +++ b/crates/renderflow-core/src/sdk.rs @@ -7,11 +7,12 @@ use std::sync::{ use serde::{Deserialize, Serialize}; use thiserror::Error; +use crate::evidence::{ArtifactManifest, DiagnosticSeverity, RunManifest}; use crate::graph::ExecutionPlan; use crate::optimization::OptimizationMode; use crate::planning::{ - execute as execute_resolved_plan, resolve as resolve_planning_request, PlanningRequest, - ResolvedExecution, + cancelled as cancelled_execution, execute as execute_resolved_plan, + resolve as resolve_planning_request, PlanningRequest, ResolvedExecution, }; use crate::toolchain::ToolchainSnapshot; @@ -157,21 +158,26 @@ pub struct ArtifactProfile { pub transforms_path: Option, } -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] -pub struct ArtifactManifest { - pub output_dir: String, - pub outputs: Vec, -} - #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct DiagnosticReport { + #[serde(default)] + pub info: Vec, + #[serde(default)] pub warnings: Vec, + #[serde(default)] pub recoverable_failures: Vec, + #[serde(default)] + pub fatal_failures: Vec, + #[serde(default)] + pub cancellations: Vec, } -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] pub struct ExecutionResult { pub manifest: ArtifactManifest, + pub run_manifest: RunManifest, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub manifest_path: Option, pub reused_cached_outputs: Vec, pub skipped_transforms: Vec, pub diagnostics: DiagnosticReport, @@ -179,6 +185,15 @@ pub struct ExecutionResult { pub toolchain: Option, } +impl ExecutionResult { + pub fn is_success(&self) -> bool { + matches!( + self.run_manifest.state, + crate::evidence::RunState::Planned | crate::evidence::RunState::Complete + ) + } +} + #[derive(Default)] pub struct EngineBuilder { reporter: Option>, @@ -306,24 +321,69 @@ impl Engine { resolved: ResolvedExecution, dry_run: bool, ) -> Result { - self.ensure_not_cancelled()?; self.emit( ProgressStage::Executing, "Executing resolved renderflow plan", ); - let result = - execute_resolved_plan(resolved, dry_run).map_err(RenderflowError::Execution)?; + let result = if self + .cancellation_token + .as_ref() + .is_some_and(CancellationToken::is_cancelled) + { + cancelled_execution(resolved).map_err(RenderflowError::Execution)? + } else { + execute_resolved_plan(resolved, dry_run).map_err(RenderflowError::Execution)? + }; self.emit(ProgressStage::Completed, "Execution complete"); + let info = result + .run_manifest + .diagnostics + .iter() + .filter(|diagnostic| diagnostic.severity == DiagnosticSeverity::Info) + .map(|diagnostic| diagnostic.message.clone()) + .collect(); + let warnings = result + .run_manifest + .diagnostics + .iter() + .filter(|diagnostic| diagnostic.severity == DiagnosticSeverity::Warning) + .map(|diagnostic| diagnostic.message.clone()) + .collect(); + let recoverable_failures = result + .run_manifest + .diagnostics + .iter() + .filter(|diagnostic| diagnostic.severity == DiagnosticSeverity::RecoverableFailure) + .map(|diagnostic| diagnostic.message.clone()) + .collect(); + let fatal_failures = result + .run_manifest + .diagnostics + .iter() + .filter(|diagnostic| diagnostic.severity == DiagnosticSeverity::FatalFailure) + .map(|diagnostic| diagnostic.message.clone()) + .collect(); + let cancellations = result + .run_manifest + .diagnostics + .iter() + .filter(|diagnostic| diagnostic.severity == DiagnosticSeverity::Cancelled) + .map(|diagnostic| diagnostic.message.clone()) + .collect(); + let reused_cached_outputs = result.run_manifest.cache_hits(); + let skipped_transforms = result.run_manifest.skipped_transforms(); Ok(ExecutionResult { - manifest: ArtifactManifest { - output_dir: result.output_dir, - outputs: result.outputs, - }, - reused_cached_outputs: Vec::new(), - skipped_transforms: Vec::new(), + manifest: result.run_manifest.artifact_manifest.clone(), + run_manifest: result.run_manifest, + manifest_path: result.manifest_path, + reused_cached_outputs, + skipped_transforms, diagnostics: DiagnosticReport { - warnings: result.diagnostics, - recoverable_failures: Vec::new(), + info, + warnings, + recoverable_failures, + fatal_failures, + cancellations, }, toolchain: result.toolchain, }) @@ -339,6 +399,7 @@ impl Engine { #[cfg(test)] mod tests { use super::*; + use crate::evidence::RunState; #[test] fn cancellation_token_reports_cancelled_state() { @@ -367,4 +428,38 @@ mod tests { assert_eq!(request.target.as_deref(), Some("html")); assert!(!request.all_targets); } + + #[test] + fn cancellation_after_planning_returns_and_persists_structured_evidence() { + let directory = tempfile::tempdir().unwrap(); + let source_path = directory.path().join("input.md"); + let output_path = directory.path().join("dist"); + let config_path = directory.path().join("renderflow.yaml"); + std::fs::write(&source_path, "# fixture\n").unwrap(); + std::fs::write( + &config_path, + format!( + "input: \"{}\"\noutput_dir: \"{}\"\noutputs:\n - type: html\n", + source_path.display(), + output_path.display() + ), + ) + .unwrap(); + + let cancellation = CancellationToken::new(); + let engine = EngineBuilder::new() + .with_cancellation_token(cancellation.clone()) + .build() + .unwrap(); + let resolved = engine + .resolve_execution(ExecutionRequest::from_path(&config_path)) + .unwrap(); + cancellation.cancel(); + + let result = engine.execute_resolved(resolved, false).unwrap(); + assert_eq!(result.run_manifest.state, RunState::Cancelled); + assert!(result.manifest.outputs.is_empty()); + assert!(!result.diagnostics.cancellations.is_empty()); + assert!(output_path.join("renderflow-run.json").is_file()); + } } diff --git a/crates/renderflow-core/tests/fixtures/execution-evidence/flow-artifact-v1.json b/crates/renderflow-core/tests/fixtures/execution-evidence/flow-artifact-v1.json new file mode 100644 index 0000000..479e402 --- /dev/null +++ b/crates/renderflow-core/tests/fixtures/execution-evidence/flow-artifact-v1.json @@ -0,0 +1,19 @@ +{ + "schema_version": "flow.artifact/v1", + "artifact_id": "artifact:sha256-abc123", + "role": "web", + "media_type": "text/html", + "digest": { + "algorithm": "sha256", + "value": "0000000000000000000000000000000000000000000000000000000000000000" + }, + "size_bytes": 42, + "producer": { + "owner": "renderflow", + "capability_id": "document.render", + "provider_version": "1.2.3" + }, + "sources": [ + "artifact:sha256-source123" + ] +} diff --git a/crates/renderflow-core/tests/fixtures/execution-evidence/outcome-matrix.json b/crates/renderflow-core/tests/fixtures/execution-evidence/outcome-matrix.json new file mode 100644 index 0000000..e4f8d33 --- /dev/null +++ b/crates/renderflow-core/tests/fixtures/execution-evidence/outcome-matrix.json @@ -0,0 +1,50 @@ +[ + { + "name": "clean_success", + "run_state": "complete", + "step_state": "complete", + "cache": "miss", + "validation": "valid", + "severity": "info" + }, + { + "name": "cache_reuse", + "run_state": "complete", + "step_state": "reused", + "cache": "hit", + "validation": "valid", + "severity": "info" + }, + { + "name": "partial_failure", + "run_state": "partial", + "step_state": "skipped", + "cache": "not_applicable", + "validation": "skipped", + "severity": "recoverable_failure" + }, + { + "name": "fatal_failure", + "run_state": "failed", + "step_state": "failed", + "cache": "miss", + "validation": "unavailable", + "severity": "fatal_failure" + }, + { + "name": "cancellation", + "run_state": "cancelled", + "step_state": "cancelled", + "cache": "not_applicable", + "validation": "skipped", + "severity": "cancelled" + }, + { + "name": "validation_failure", + "run_state": "failed", + "step_state": "failed", + "cache": "miss", + "validation": "invalid", + "severity": "fatal_failure" + } +] diff --git a/docs/artifact-kernel.md b/docs/artifact-kernel.md index 444b96f..42b93b3 100644 --- a/docs/artifact-kernel.md +++ b/docs/artifact-kernel.md @@ -64,7 +64,7 @@ This removes the source and final-write UTF-8 assumptions from graph execution. ## 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. +The kernel deliberately does not depend on `egohygiene/flow`. The [execution evidence](execution-evidence.md) layer records provenance-complete native run and artifact manifests and provides an explicit projection into the currently supported `flow.artifact/v1` interchange shape. ## Migration sequence diff --git a/docs/execution-evidence.md b/docs/execution-evidence.md new file mode 100644 index 0000000..dc9654c --- /dev/null +++ b/docs/execution-evidence.md @@ -0,0 +1,35 @@ +# Execution evidence + +Every non-dry canonical build writes a versioned `renderflow-run.json` manifest in the configured output directory. The manifest is authoritative for what the executor actually produced, reused, skipped, validated, or failed to produce; it is not reconstructed from requested targets. + +Dry runs return the same evidence type with `state: planned`, but remain side-effect free and do not persist a manifest. + +## Outcome states + +The top-level `state` is one of: + +- `complete`: every selected target was produced and materialized; +- `partial`: at least one selected output was materialized and another step or output failed; +- `failed`: no selected output was materialized successfully; +- `cancelled`: execution was cancelled after planning and before transforms started; +- `planned`: dry-run evidence only. + +The CLI exits unsuccessfully for `partial`, `failed`, and `cancelled` outcomes after reporting the manifest path. SDK callers receive the structured `ExecutionResult` and should inspect `run_manifest.state`. + +## Artifact and step evidence + +The artifact manifest contains source, retained intermediate, and terminal artifact records. Each record includes a stable artifact ID, logical role, lifecycle, safe store or bundle locator, canonical format and media type, SHA-256 digest, size, producer identity, source lineage, cache status, validation status, and fidelity declaration. + +Each executed DAG edge produces step evidence with transform/capability/provider identity, input and output artifact IDs, a configuration digest, timestamps, duration, cache disposition, validation and fidelity states, and structured diagnostics. Cache hits use `state: reused`; transforms blocked by a failed dependency use `state: skipped` with a reason. + +The machine-readable contract is [`schemas/renderflow-run-v1.schema.json`](https://github.com/egohygiene/renderflow/blob/main/schemas/renderflow-run-v1.schema.json). + +## Flow compatibility + +`RunManifest::flow_artifacts_v1()` explicitly projects native artifact evidence into Flow's provisional `flow.artifact/v1` interchange shape. Renderflow keeps its richer native evidence independent from Flow and pins a compatibility fixture to the contract present in `egohygiene/flow` commit `a7d28ee812f9d6ccd93b6786be924e24c455accd`. + +Native IDs such as `artifact:sha256:` are mapped deterministically to Flow-compatible IDs such as `artifact:sha256-`. Producer fields are projected as `owner`, `capability_id`, and `provider_version`. + +## Sensitive data boundary + +Run manifests contain digests of the resolved plan and source spec, not serialized configuration or environment variables. Step configuration is represented only by a SHA-256 digest. Artifact locators are relative `artifact-store:` or `bundle:` locators. Provider diagnostics are retained for operability, so provider implementations must not place credentials or secret values in error messages. diff --git a/mkdocs.yml b/mkdocs.yml index 1d91dab..3dc9243 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -110,6 +110,7 @@ nav: - Implementation Views: - Graph Engine: architecture/graph-engine.md - DAG Execution: architecture/dag-execution.md + - Execution Evidence: execution-evidence.md - Plugin Architecture: architecture/plugin-architecture.md - Execution Plans: architecture/execution-plans.md - CLI Reference: diff --git a/schemas/renderflow-run-v1.schema.json b/schemas/renderflow-run-v1.schema.json new file mode 100644 index 0000000..f2b7978 --- /dev/null +++ b/schemas/renderflow-run-v1.schema.json @@ -0,0 +1,214 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://egohygiene.github.io/renderflow/schemas/renderflow-run-v1.schema.json", + "title": "Renderflow run manifest v1", + "type": "object", + "additionalProperties": false, + "required": [ + "schema_version", + "run_id", + "execution_plan_digest", + "source_spec_digest", + "engine_version", + "started_at_unix_ms", + "completed_at_unix_ms", + "state", + "artifact_manifest", + "steps", + "diagnostics" + ], + "properties": { + "schema_version": { "const": "renderflow.run/v1" }, + "run_id": { "type": "string", "pattern": "^run:sha256:[a-f0-9]{64}$" }, + "execution_plan_digest": { "$ref": "#/$defs/digest" }, + "source_spec_digest": { "$ref": "#/$defs/digest" }, + "engine_version": { "type": "string", "minLength": 1 }, + "started_at_unix_ms": { "type": "integer", "minimum": 0 }, + "completed_at_unix_ms": { "type": "integer", "minimum": 0 }, + "state": { + "enum": ["planned", "complete", "partial", "failed", "cancelled"] + }, + "artifact_manifest": { "$ref": "#/$defs/artifact_manifest" }, + "steps": { + "type": "array", + "items": { "$ref": "#/$defs/step" } + }, + "diagnostics": { + "type": "array", + "items": { "$ref": "#/$defs/diagnostic" } + }, + "toolchain": { "type": "object" } + }, + "$defs": { + "digest": { + "type": "object", + "additionalProperties": false, + "required": ["algorithm", "value"], + "properties": { + "algorithm": { "const": "sha256" }, + "value": { "type": "string", "pattern": "^[a-f0-9]{64}$" } + } + }, + "producer": { + "type": "object", + "additionalProperties": false, + "required": ["system"], + "properties": { + "system": { "type": "string", "minLength": 1 }, + "transform": { "type": "string", "minLength": 1 }, + "capability": { "type": "string", "minLength": 1 }, + "provider": { "type": "string", "minLength": 1 }, + "version": { "type": "string", "minLength": 1 } + } + }, + "artifact": { + "type": "object", + "additionalProperties": false, + "required": [ + "artifact_id", + "role", + "lifecycle", + "locator", + "format", + "media_type", + "digest", + "size_bytes", + "producer", + "sources", + "cache", + "validation", + "fidelity" + ], + "properties": { + "artifact_id": { "type": "string", "minLength": 1 }, + "role": { "type": "string", "minLength": 1 }, + "lifecycle": { "enum": ["source", "intermediate", "terminal"] }, + "locator": { "type": "string", "minLength": 1 }, + "format": { "type": "string", "minLength": 1 }, + "media_type": { "type": "string", "pattern": "^[^/]+/[^/]+$" }, + "digest": { "$ref": "#/$defs/digest" }, + "size_bytes": { "type": "integer", "minimum": 0 }, + "producer": { "$ref": "#/$defs/producer" }, + "sources": { + "type": "array", + "uniqueItems": true, + "items": { "type": "string", "minLength": 1 } + }, + "cache": { "enum": ["source", "miss", "hit", "not_applicable"] }, + "validation": { + "enum": [ + "valid", + "valid_with_warnings", + "invalid", + "unavailable", + "skipped", + "not_requested" + ] + }, + "fidelity": { + "enum": ["lossless", "partial", "lossy", "path_dependent", "unknown"] + }, + "warnings": { "type": "array", "items": { "type": "string" } }, + "metadata": { "type": "object" } + } + }, + "artifact_manifest": { + "type": "object", + "additionalProperties": false, + "required": [ + "schema_version", + "run_id", + "output_dir", + "outputs", + "artifacts" + ], + "properties": { + "schema_version": { "const": "renderflow.artifact-manifest/v1" }, + "run_id": { "type": "string", "pattern": "^run:sha256:[a-f0-9]{64}$" }, + "output_dir": { "type": "string" }, + "outputs": { "type": "array", "items": { "type": "string" } }, + "artifacts": { + "type": "array", + "items": { "$ref": "#/$defs/artifact" } + } + } + }, + "step": { + "type": "object", + "additionalProperties": false, + "required": [ + "step_id", + "transform", + "transform_version", + "input_artifacts", + "output_artifacts", + "configuration_digest", + "started_at_unix_ms", + "completed_at_unix_ms", + "duration_ms", + "state", + "cache", + "validation", + "fidelity" + ], + "properties": { + "step_id": { "type": "string", "minLength": 1 }, + "transform": { "type": "string", "minLength": 1 }, + "transform_version": { "type": "string", "minLength": 1 }, + "capability": { "type": "string", "minLength": 1 }, + "provider": { "type": "string", "minLength": 1 }, + "input_artifacts": { + "type": "array", + "items": { "type": "string", "minLength": 1 } + }, + "output_artifacts": { + "type": "array", + "items": { "type": "string", "minLength": 1 } + }, + "configuration_digest": { "$ref": "#/$defs/digest" }, + "started_at_unix_ms": { "type": "integer", "minimum": 0 }, + "completed_at_unix_ms": { "type": "integer", "minimum": 0 }, + "duration_ms": { "type": "integer", "minimum": 0 }, + "state": { "enum": ["complete", "reused", "skipped", "failed", "cancelled"] }, + "cache": { "enum": ["source", "miss", "hit", "not_applicable"] }, + "validation": { + "enum": [ + "valid", + "valid_with_warnings", + "invalid", + "unavailable", + "skipped", + "not_requested" + ] + }, + "fidelity": { + "enum": ["lossless", "partial", "lossy", "path_dependent", "unknown"] + }, + "skip_reason": { "type": "string", "minLength": 1 }, + "diagnostics": { + "type": "array", + "items": { "$ref": "#/$defs/diagnostic" } + } + } + }, + "diagnostic": { + "type": "object", + "additionalProperties": false, + "required": ["severity", "code", "message"], + "properties": { + "severity": { + "enum": [ + "info", + "warning", + "recoverable_failure", + "fatal_failure", + "cancelled" + ] + }, + "code": { "type": "string", "minLength": 1 }, + "message": { "type": "string" }, + "step_id": { "type": "string", "minLength": 1 } + } + } + } +}