diff --git a/crates/attractor/src/artifact.rs b/crates/attractor/src/artifact.rs index 37bc752bf..6ecd9b3a5 100644 --- a/crates/attractor/src/artifact.rs +++ b/crates/attractor/src/artifact.rs @@ -6,6 +6,8 @@ use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use serde_json::Value; +use agent::ExecutionEnvironment; + use crate::error::{AttractorError, Result}; /// Threshold above which artifacts are stored on disk instead of in memory (100KB). @@ -194,12 +196,12 @@ impl ArtifactStore { } /// Prefix used to identify artifact pointer strings in context values. -const ARTIFACT_POINTER_PREFIX: &str = "artifact://"; +const ARTIFACT_POINTER_PREFIX: &str = "file://"; /// Offload context values exceeding the file-backing threshold into the artifact store. /// /// For each entry in `updates` whose serialized JSON exceeds `FILE_BACKING_THRESHOLD`, -/// the value is stored in `store` and replaced with an `"artifact://{path}"` pointer. +/// the value is stored in `store` and replaced with a `"file://{path}"` pointer. /// Small values are left untouched. /// /// # Errors @@ -225,7 +227,7 @@ pub fn offload_large_values( /// Extract the file path from an artifact pointer value. /// -/// Returns `Some(path)` if the value is a string starting with `"artifact://"`, +/// Returns `Some(path)` if the value is a string starting with `"file://"`, /// `None` otherwise. #[must_use] pub fn artifact_path(value: &Value) -> Option<&str> { @@ -234,7 +236,7 @@ pub fn artifact_path(value: &Value) -> Option<&str> { .and_then(|s| s.strip_prefix(ARTIFACT_POINTER_PREFIX)) } -/// Returns `true` if `path` looks like an artifact pointer path (starts with `"artifact://"`). +/// Returns `true` if `path` looks like an artifact pointer path (starts with `"file://"`). #[must_use] pub fn is_artifact_pointer(value: &Value) -> bool { artifact_path(value).is_some() @@ -242,12 +244,65 @@ pub fn is_artifact_pointer(value: &Value) -> bool { /// Resolve an artifact pointer to the base name displayed in preamble rendering. /// -/// Given `"artifact:///tmp/logs/artifacts/response.plan.json"`, returns `"See: /tmp/logs/artifacts/response.plan.json"`. +/// Given `"file:///tmp/logs/artifacts/response.plan.json"`, returns `"See: /tmp/logs/artifacts/response.plan.json"`. #[must_use] pub fn format_artifact_reference(path: &str) -> String { format!("See: {path}") } +/// Sync artifact files to a remote execution environment. +/// +/// For each `file://` pointer in `updates`, checks whether the file is accessible +/// in `env`. If not, reads the local file and uploads it via `env.write_file`, +/// placing it at `{working_directory}/.attractor/artifacts/{filename}`. The pointer +/// is rewritten to reference the remote path. +/// +/// # Errors +/// +/// Returns an error if reading a local artifact or writing to the remote env fails. +pub async fn sync_artifacts_to_env( + updates: &mut HashMap, + env: &dyn ExecutionEnvironment, +) -> Result<()> { + for value in updates.values_mut() { + let local_path = match artifact_path(value) { + Some(p) => p.to_string(), + None => continue, + }; + + match env.file_exists(&local_path).await { + Ok(true) => continue, + Ok(false) => {} + Err(e) => { + return Err(AttractorError::Engine(format!( + "failed to check artifact existence: {e}" + ))); + } + } + + let content = std::fs::read_to_string(&local_path).map_err(|e| { + AttractorError::Engine(format!("failed to read local artifact {local_path}: {e}")) + })?; + + let filename = std::path::Path::new(&local_path) + .file_name() + .and_then(|f| f.to_str()) + .unwrap_or("artifact.json"); + + let remote_path = format!( + "{}/.attractor/artifacts/{filename}", + env.working_directory() + ); + + env.write_file(&remote_path, &content).await.map_err(|e| { + AttractorError::Engine(format!("failed to write artifact to remote env: {e}")) + })?; + + *value = Value::String(format!("{ARTIFACT_POINTER_PREFIX}{remote_path}")); + } + Ok(()) +} + #[cfg(test)] mod tests { use super::*; @@ -415,7 +470,7 @@ mod tests { #[test] fn artifact_path_extracts_path_from_pointer() { - let value = serde_json::json!("artifact:///tmp/logs/artifacts/response.plan.json"); + let value = serde_json::json!("file:///tmp/logs/artifacts/response.plan.json"); assert_eq!( artifact_path(&value), Some("/tmp/logs/artifacts/response.plan.json") @@ -433,4 +488,153 @@ mod tests { let value = serde_json::json!(42); assert_eq!(artifact_path(&value), None); } + + // --- sync_artifacts_to_env tests --- + + use std::sync::Mutex; + + struct TestSyncEnv { + accessible: bool, + written: Mutex>, + working_dir: String, + } + + impl TestSyncEnv { + fn new(accessible: bool, working_dir: &str) -> Self { + Self { + accessible, + written: Mutex::new(Vec::new()), + working_dir: working_dir.to_string(), + } + } + } + + #[async_trait::async_trait] + impl ExecutionEnvironment for TestSyncEnv { + async fn read_file(&self, _path: &str, _offset: Option, _limit: Option) -> std::result::Result { + Err("not implemented".to_string()) + } + + async fn write_file(&self, path: &str, content: &str) -> std::result::Result<(), String> { + self.written + .lock() + .unwrap() + .push((path.to_string(), content.to_string())); + Ok(()) + } + + async fn delete_file(&self, _path: &str) -> std::result::Result<(), String> { + Err("not implemented".to_string()) + } + + async fn file_exists(&self, _path: &str) -> std::result::Result { + Ok(self.accessible) + } + + async fn list_directory(&self, _path: &str, _depth: Option) -> std::result::Result, String> { + Err("not implemented".to_string()) + } + + async fn exec_command( + &self, + _command: &str, + _timeout_ms: u64, + _working_dir: Option<&str>, + _env_vars: Option<&std::collections::HashMap>, + _cancel_token: Option, + ) -> std::result::Result { + Err("not implemented".to_string()) + } + + async fn grep(&self, _pattern: &str, _path: &str, _options: &agent::GrepOptions) -> std::result::Result, String> { + Err("not implemented".to_string()) + } + + async fn glob(&self, _pattern: &str, _path: Option<&str>) -> std::result::Result, String> { + Err("not implemented".to_string()) + } + + async fn initialize(&self) -> std::result::Result<(), String> { + Ok(()) + } + + async fn cleanup(&self) -> std::result::Result<(), String> { + Ok(()) + } + + fn working_directory(&self) -> &str { + &self.working_dir + } + + fn platform(&self) -> &str { + "linux" + } + + fn os_version(&self) -> String { + "Linux 5.15".to_string() + } + } + + #[tokio::test] + async fn sync_uploads_artifact_when_not_accessible() { + let dir = tempfile::tempdir().unwrap(); + let artifact_file = dir.path().join("response.plan.json"); + std::fs::write(&artifact_file, r#""hello from artifact""#).unwrap(); + + let pointer = format!("file://{}", artifact_file.display()); + let mut updates = HashMap::new(); + updates.insert("response.plan".to_string(), Value::String(pointer)); + + let env = TestSyncEnv::new(false, "/workspace"); + sync_artifacts_to_env(&mut updates, &env).await.unwrap(); + + let written = env.written.lock().unwrap(); + assert_eq!(written.len(), 1); + assert_eq!( + written[0].0, + "/workspace/.attractor/artifacts/response.plan.json" + ); + assert_eq!(written[0].1, r#""hello from artifact""#); + + let new_pointer = updates["response.plan"].as_str().unwrap(); + assert_eq!( + new_pointer, + "file:///workspace/.attractor/artifacts/response.plan.json" + ); + } + + #[tokio::test] + async fn sync_skips_when_artifact_already_accessible() { + let dir = tempfile::tempdir().unwrap(); + let artifact_file = dir.path().join("data.json"); + std::fs::write(&artifact_file, "{}").unwrap(); + + let pointer = format!("file://{}", artifact_file.display()); + let mut updates = HashMap::new(); + updates.insert("key".to_string(), Value::String(pointer.clone())); + + let env = TestSyncEnv::new(true, "/workspace"); + sync_artifacts_to_env(&mut updates, &env).await.unwrap(); + + let written = env.written.lock().unwrap(); + assert!(written.is_empty()); + assert_eq!(updates["key"].as_str().unwrap(), &pointer); + } + + #[tokio::test] + async fn sync_ignores_non_artifact_values() { + let mut updates = HashMap::new(); + updates.insert("name".to_string(), serde_json::json!("Alice")); + updates.insert("count".to_string(), serde_json::json!(42)); + updates.insert("nested".to_string(), serde_json::json!({"a": 1})); + + let env = TestSyncEnv::new(false, "/workspace"); + sync_artifacts_to_env(&mut updates, &env).await.unwrap(); + + let written = env.written.lock().unwrap(); + assert!(written.is_empty()); + assert_eq!(updates["name"], serde_json::json!("Alice")); + assert_eq!(updates["count"], serde_json::json!(42)); + assert_eq!(updates["nested"], serde_json::json!({"a": 1})); + } } diff --git a/crates/attractor/src/engine.rs b/crates/attractor/src/engine.rs index 4708f564a..2058ff86d 100644 --- a/crates/attractor/src/engine.rs +++ b/crates/attractor/src/engine.rs @@ -10,7 +10,7 @@ use chrono::Utc; use futures::FutureExt; use rand::Rng; -use crate::artifact::{offload_large_values, ArtifactStore}; +use crate::artifact::{offload_large_values, sync_artifacts_to_env, ArtifactStore}; use crate::checkpoint::Checkpoint; use crate::condition::evaluate_condition; use crate::context::Context; @@ -935,6 +935,14 @@ impl PipelineEngine { context.append_log(format!("artifact offload failed: {e}")); } + // Sync artifact files to the execution environment (no-op for local envs) + if let Err(e) = sync_artifacts_to_env( + &mut outcome.context_updates, + &*self.services.execution_env, + ).await { + context.append_log(format!("artifact sync failed: {e}")); + } + // Step 3: Record completion completed_nodes.push(node.id.clone()); node_outcomes.insert(node.id.clone(), outcome.clone()); diff --git a/crates/attractor/tests/integration.rs b/crates/attractor/tests/integration.rs index 34095a23e..814e8f4b6 100644 --- a/crates/attractor/tests/integration.rs +++ b/crates/attractor/tests/integration.rs @@ -6899,7 +6899,7 @@ async fn large_context_values_are_offloaded_to_artifact_store() { .expect("context should have response.big_output"); let pointer_str = pointer_value.as_str().expect("pointer should be a string"); assert!( - pointer_str.starts_with("artifact://"), + pointer_str.starts_with("file://"), "value should be an artifact pointer, got: {pointer_str}" ); @@ -6933,4 +6933,153 @@ async fn large_context_values_are_offloaded_to_artifact_store() { "artifact_count should be > 0, got {artifact_count}" ); } +} + +// --------------------------------------------------------------------------- +// Artifact sync to remote execution environments +// --------------------------------------------------------------------------- + +/// A mock execution environment where `file_exists` always returns false, +/// simulating a remote container that doesn't have local artifact files. +struct RemoteMockEnv { + working_dir: String, + written: std::sync::Mutex>, +} + +impl RemoteMockEnv { + fn new(working_dir: &str) -> Self { + Self { + working_dir: working_dir.to_string(), + written: std::sync::Mutex::new(Vec::new()), + } + } +} + +#[async_trait::async_trait] +impl agent::ExecutionEnvironment for RemoteMockEnv { + async fn read_file(&self, _path: &str, _offset: Option, _limit: Option) -> std::result::Result { + Err("not implemented".to_string()) + } + + async fn write_file(&self, path: &str, content: &str) -> std::result::Result<(), String> { + self.written + .lock() + .unwrap() + .push((path.to_string(), content.to_string())); + Ok(()) + } + + async fn delete_file(&self, _path: &str) -> std::result::Result<(), String> { + Err("not implemented".to_string()) + } + + async fn file_exists(&self, _path: &str) -> std::result::Result { + Ok(false) + } + + async fn list_directory(&self, _path: &str, _depth: Option) -> std::result::Result, String> { + Err("not implemented".to_string()) + } + + async fn exec_command( + &self, + _command: &str, + _timeout_ms: u64, + _working_dir: Option<&str>, + _env_vars: Option<&std::collections::HashMap>, + _cancel_token: Option, + ) -> std::result::Result { + Err("not implemented".to_string()) + } + + async fn grep(&self, _pattern: &str, _path: &str, _options: &agent::GrepOptions) -> std::result::Result, String> { + Err("not implemented".to_string()) + } + + async fn glob(&self, _pattern: &str, _path: Option<&str>) -> std::result::Result, String> { + Err("not implemented".to_string()) + } + + async fn initialize(&self) -> std::result::Result<(), String> { + Ok(()) + } + + async fn cleanup(&self) -> std::result::Result<(), String> { + Ok(()) + } + + fn working_directory(&self) -> &str { + &self.working_dir + } + + fn platform(&self) -> &str { + "linux" + } + + fn os_version(&self) -> String { + "Linux 5.15".to_string() + } +} + +#[tokio::test] +async fn artifact_pointers_rewritten_for_remote_execution_env() { + // Pipeline: start -> big_output -> exit + // big_output uses LargeOutputHandler which returns a >100KB context_update. + // RemoteMockEnv simulates a container where local files don't exist. + let mut graph = make_graph_with_start_exit("ArtifactSync"); + graph.attrs.insert( + "goal".to_string(), + AttrValue::String("Test artifact sync to remote env".to_string()), + ); + + let mut big_output = Node::new("big_output"); + big_output.attrs.insert( + "label".to_string(), + AttrValue::String("Big Output".to_string()), + ); + graph.nodes.insert("big_output".to_string(), big_output); + + graph.edges.push(Edge::new("start", "big_output")); + graph.edges.push(Edge::new("big_output", "exit")); + + let dir = tempfile::tempdir().unwrap(); + let mut registry = HandlerRegistry::new(Box::new(LargeOutputHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + + let remote_env = Arc::new(RemoteMockEnv::new("/sandbox")); + let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), remote_env.clone()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + dry_run: false, + }; + + let outcome = engine + .run(&graph, &config) + .await + .expect("pipeline should succeed"); + assert_eq!(outcome.status, StageStatus::Success); + + // The checkpoint context should contain a pointer rewritten for the remote env + let checkpoint = Checkpoint::load(&dir.path().join("checkpoint.json")) + .expect("checkpoint should load"); + let pointer_value = checkpoint + .context_values + .get("response.big_output") + .expect("context should have response.big_output"); + let pointer_str = pointer_value.as_str().expect("pointer should be a string"); + assert!( + pointer_str.starts_with("file:///sandbox/.attractor/artifacts/"), + "pointer should reference remote path, got: {pointer_str}" + ); + + // The RemoteMockEnv should have received exactly one write with >100KB content + let written = remote_env.written.lock().unwrap(); + assert_eq!(written.len(), 1, "should have written 1 artifact"); + assert!( + written[0].1.len() > 100 * 1024, + "written content should be >100KB, got {} bytes", + written[0].1.len() + ); } \ No newline at end of file