diff --git a/docs-internal/run-directory-keys.md b/docs-internal/run-directory-keys.md index 04c9e5eba..4d5adc8e7 100644 --- a/docs-internal/run-directory-keys.md +++ b/docs-internal/run-directory-keys.md @@ -26,7 +26,6 @@ These paths are local runtime state, not canonical event projections. | `worktree/` | Git worktree used by checkpointed runs | | `runtime/blobs/` | Materialized local blob payloads for file-backed `fabro+blob://` references | | `runtime/worker.stderr.log` | Server-managed worker stderr capture | -| `cache/artifacts/files/` | Captured artifact files organized by node and retry | | `nodes/{manager_node}_{visit}/child/` | Nested scratch root for manager-loop child workflows | ## Reconstructed / Exported Files diff --git a/docs/brainstorms/2026-04-08-eliminate-scratch-artifact-cache-requirements.md b/docs/brainstorms/2026-04-08-eliminate-scratch-artifact-cache-requirements.md new file mode 100644 index 000000000..6e9aea286 --- /dev/null +++ b/docs/brainstorms/2026-04-08-eliminate-scratch-artifact-cache-requirements.md @@ -0,0 +1,69 @@ +--- +date: 2026-04-08 +topic: eliminate-scratch-artifact-cache +--- + +# Eliminate Scratch Artifact Cache + +## Problem Frame + +The artifact collection pipeline uses `scratch/cache/artifacts/files/` as a staging area for CLI uploads: files are downloaded from the sandbox to local disk, hashed, then uploaded to the server's `ArtifactStore`. Now that `ArtifactStore` (backed by object store) is the durable source of truth, this persistent local cache is unnecessary overhead. It adds disk usage and complicates the scratch directory contract. It also obscures a separate bug: server-managed runs currently start `ArtifactLifecycle` with no artifact sink configured, so collected artifacts are discarded. Eliminating the cache alone does not fix that bug; server-managed runs also need a direct `ArtifactStore` write path. + +## Requirements + +**Pipeline: Replace persistent cache with transient local staging** + +- R1. `collect_artifacts` must download each file from the sandbox into a transient per-attempt local directory tree that preserves each artifact's relative path, compute MD5/SHA256 from those transient local files, and keep that transient tree available until the configured artifact sink has finished consuming it, including upload retries. Delete the transient tree immediately after direct store writes or HTTP uploads finish. No persistent `cache/artifacts/` directory. +- R2. `CapturedArtifactInfo` (path, mime, hashes, bytes) must still be produced for each collected file and emitted via `ArtifactCaptured` events. + +**Store: Write artifacts directly to ArtifactStore** + +- R3. When running server-side, `ArtifactLifecycle` writes collected artifacts directly to the server's `ArtifactStore`, fixing the current gap where server-managed runs configure no artifact sink and silently discard artifacts. +- R4. When running via CLI, the existing HTTP upload path (`HttpArtifactUploader`) continues to work. Internal uploader abstractions may change as needed, but the CLI must continue uploading artifacts to the server rather than writing directly to `ArtifactStore`. +- R5. `ArtifactLifecycle` must be configured with exactly one artifact sink for artifact-enabled runs: either a direct `ArtifactStore`-backed sink for server-managed runs or an HTTP uploader-backed sink for CLI runs. It must also hold a `RunId` for direct `ArtifactStore::put` calls. The internal representation of that sink (enum, unified trait, or refined existing trait) is deferred to planning. + +**Scratch cleanup** + +- R6. Remove `artifact_cache_dir()`, `artifact_files_dir()`, and `artifact_stage_dir()` from `RunScratch` in `fabro-config/src/storage.rs`. +- R7. Remove the `create_dir_all(self.artifact_files_dir())` from `RunScratch::create()`. +- R8. Update `docs/reference/run-directory.mdx` and `docs-internal/run-directory-keys.md` to remove all `cache/artifacts/` entries (including `cache/artifacts/files/`). +- R10. Update integration tests in `fabro-workflow/tests/it/integration.rs` and `daytona_integration.rs` that assert on `artifact_stage_dir` paths to verify artifacts via `ArtifactStore` or uploader behavior instead of local filesystem paths. + +**CLI output** + +- R9. The post-run "=== Artifacts ===" output must stop printing local scratch paths. It should list durable artifact identifiers derived from stored metadata, at minimum `node_slug`, `retry`, and `relative_path`, and reference `fabro artifact cp` for retrieval. +- R11. Removing `artifact_stage_dir()` must not change durable per-stage/per-attempt grouping. Artifacts must still be addressable by `StageId`, and CLI/server artifact surfaces must continue exposing `node_slug`, `retry`, and `relative_path` from store-backed metadata rather than reconstructing local scratch paths. + +## Success Criteria + +- Server-managed workflow runs produce artifacts in `ArtifactStore` (previously they did not). +- No `cache/artifacts/` directory is created or written to during any workflow run. +- `fabro artifact list` and `fabro artifact cp` continue to work unchanged. +- The post-run artifact summary prints logical artifact identifiers (`node_slug`, `retry`, `relative_path`) instead of local scratch paths. +- `ArtifactCaptured` events still contain correct hashes and byte counts. + +## Scope Boundaries + +- No changes to `fabro artifact list` or `fabro artifact cp` commands. +- No changes to the `Sandbox` trait (no streaming download API). +- No changes to the HTTP artifact upload protocol between CLI and server. +- `runtime/blobs/` (context value materialization) is a separate concern, unchanged. + +## Key Decisions + +- **Transient local staging, not persistent scratch cache**: The `Sandbox` trait only has `download_file_to_local`. Rather than adding a streaming API to all sandbox impls, use transient local files outside `RunScratch`, but keep them alive until the configured sink has finished consuming them. +- **Server writes to ArtifactStore directly**: The server already owns the `ArtifactStore` instance. Server-managed runs should use that directly instead of relying on an uploader being present. +- **CLI keeps HTTP upload path**: The CLI continues uploading artifacts to the server over HTTP. The internal uploader interface may change, but the wire protocol stays the same. Smaller blast radius than giving the CLI its own local `ArtifactStore`. + +## Outstanding Questions + +### Deferred to Planning + +- [Affects R5][Technical] What internal representation should `ArtifactLifecycle` use for the exactly-one artifact sink? Options: an enum with server/CLI variants, a new unified trait, or a refactor of the existing uploader trait. +- [Affects R1][Technical] Should `collect_artifacts` return transient local handles/paths alongside `CapturedArtifactInfo`, or should store/upload happen inside the collection step while the transient files are definitely still present? +- [Affects R1][Technical] Should transient staging use `tempfile::TempDir` (auto-cleanup on drop) or manual `std::fs::remove_dir_all`? The former is more robust against panics. +- [Affects R9][Needs research] What does the current CLI output look like for artifacts, and what is the best replacement format? Check `lib/crates/fabro-cli/src/commands/run/output.rs`. + +## Next Steps + +-> `/ce:plan` for structured implementation planning diff --git a/docs/reference/run-directory.mdx b/docs/reference/run-directory.mdx index ad99f7e5f..6bab5df6a 100644 --- a/docs/reference/run-directory.mdx +++ b/docs/reference/run-directory.mdx @@ -28,7 +28,6 @@ These paths are local runtime state and caches, not the canonical run record. - **`worktree/`** — When running in worktree mode, Fabro creates a Git worktree here as the working directory for agents and commands. - **`runtime/`** — Local runtime files. Today this is mainly materialized blob payloads under `runtime/blobs/`. -- **`cache/artifacts/files/`** — Captured artifact files organized by node and retry. - **`nodes/{manager_node}_{visit}/child/`** — Nested scratch directories for manager-loop child workflows. Large durable values, event streams, checkpoints, diffs, conclusions, and retros are no longer projected into live scratch by default. Use `fabro logs`, `fabro inspect`, the API, or `fabro store dump` for those surfaces. diff --git a/lib/crates/fabro-cli/src/commands/run/output.rs b/lib/crates/fabro-cli/src/commands/run/output.rs index 1372d38ac..43ac31d10 100644 --- a/lib/crates/fabro-cli/src/commands/run/output.rs +++ b/lib/crates/fabro-cli/src/commands/run/output.rs @@ -1,9 +1,8 @@ -use std::path::{Path, PathBuf}; +use std::path::Path; use std::time::Duration; use anyhow::{Context as _, Result}; use fabro_api::types; -use fabro_config::RunScratch; use fabro_types::{ PullRequestRecord, RunBlobId, RunId, parse_blob_ref, parse_legacy_blob_file_ref, }; @@ -153,8 +152,8 @@ pub(crate) async fn print_run_summary_with_client( let final_output = resolve_final_output_with_client(client, run_id, checkpoint.as_ref()).await?; print_final_output(final_output.as_deref(), styles); - if let Some(run_dir) = local_run_dir { - print_assets_with_client(client, run_id, run_dir, styles).await?; + if local_run_dir.is_some() { + print_assets_with_client(client, run_id, styles).await?; } Ok(()) } @@ -318,49 +317,54 @@ fn blob_id_from_response(response: &str) -> Option { parse_blob_ref(response).or_else(|| parse_legacy_blob_file_ref(response)) } -async fn resolve_local_artifact_display_paths_with_client( +async fn list_artifact_display_entries_with_client( client: &server_client::ServerStoreClient, run_id: &RunId, - run_dir: &Path, -) -> Result> { - let mut paths = Vec::new(); +) -> Result> { + let mut entries = Vec::new(); for entry in client.list_run_artifacts(run_id).await? { let retry = u32::try_from(entry.retry) .context("server returned invalid negative artifact retry")?; - let path = RunScratch::new(run_dir) - .artifact_stage_dir(&entry.node_slug, retry) - .join(entry.relative_path); - paths.push(path); + entries.push((entry.node_slug, retry, entry.relative_path)); } - paths.sort(); - Ok(paths) + entries.sort(); + Ok(entries) } async fn print_assets_with_client( client: &server_client::ServerStoreClient, run_id: &RunId, - run_dir: &Path, styles: &Styles, ) -> Result<()> { - let paths = resolve_local_artifact_display_paths_with_client(client, run_id, run_dir).await?; - if paths.is_empty() { + let entries = list_artifact_display_entries_with_client(client, run_id).await?; + if entries.is_empty() { return Ok(()); } - let home = dirs::home_dir(); + + let node_width = entries + .iter() + .map(|(node_slug, _, _)| node_slug.len()) + .max() + .unwrap_or(4) + .max(4); + let retry_width = entries + .iter() + .map(|(_, retry, _)| retry.to_string().len()) + .max() + .unwrap_or(5) + .max(5); + eprintln!("\n{}", styles.bold.apply_to("=== Artifacts ===")); - for path in &paths { - let display = match &home { - Some(home_dir) => { - let home_str = home_dir.to_string_lossy(); - if let Some(rest) = path.to_string_lossy().strip_prefix(home_str.as_ref()) { - format!("~{rest}") - } else { - path.display().to_string() - } - } - None => path.display().to_string(), - }; - eprintln!("{display}"); + eprintln!("{:retry_width$} PATH", "NODE", "RETRY"); + for (node_slug, retry, relative_path) in &entries { + eprintln!("{node_slug:retry_width$} {relative_path}"); } + eprintln!(); + eprintln!( + "{}", + styles.dim.apply_to(format!( + "Copy with: fabro artifact cp {run_id}: --node --retry " + )) + ); Ok(()) } diff --git a/lib/crates/fabro-cli/src/commands/run/runner.rs b/lib/crates/fabro-cli/src/commands/run/runner.rs index a7c64b0fd..c9ec4d919 100644 --- a/lib/crates/fabro-cli/src/commands/run/runner.rs +++ b/lib/crates/fabro-cli/src/commands/run/runner.rs @@ -9,7 +9,7 @@ use fabro_interview::{ControlInterviewer, WorkerControlEnvelope, WorkerControlMe use fabro_store::{EventEnvelope, EventPayload, RunProjection}; use fabro_types::{EventBody, RunBlobId, RunEvent, RunId, Settings, StatusReason}; use fabro_workflow::artifact_snapshot::CapturedArtifactInfo; -use fabro_workflow::artifact_upload::StageArtifactUploader; +use fabro_workflow::artifact_upload::{ArtifactSink, StageArtifactUploader}; use fabro_workflow::event::{Emitter, RunEventSink}; use fabro_workflow::operations::{self, StartServices}; use fabro_workflow::run_control::RunControlState; @@ -63,11 +63,11 @@ pub(crate) async fn execute( .run .as_ref() .ok_or_else(|| anyhow!("Run {run_id} has no run record in store"))?; - let artifact_uploader = Some(build_artifact_uploader( + let artifact_sink = Some(ArtifactSink::Uploader(build_artifact_uploader( run_id, client.clone_for_reuse(), artifact_upload_token, - )); + ))); let interviewer = Arc::new(ControlInterviewer::new()); let cancel_token = Arc::new(AtomicBool::new(false)); tokio::spawn(read_worker_control_stream( @@ -91,7 +91,7 @@ pub(crate) async fn execute( async move { Ok(()) } }), ]), - artifact_uploader, + artifact_sink, run_control: Some(run_control), github_app, on_node: None, diff --git a/lib/crates/fabro-cli/tests/it/cmd/run.rs b/lib/crates/fabro-cli/tests/it/cmd/run.rs index 76e94c57b..ce90bd159 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/run.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/run.rs @@ -496,14 +496,11 @@ include = ["assets/**"] ); let stderr = output_stderr(&output); - let expected_path = context - .find_run_dir(&run_id) - .join("cache/artifacts/files/create_assets/retry_1/assets/shared/report.txt"); assert!(stderr.contains("=== Artifacts ==="), "{stderr}"); - assert!( - stderr.contains(expected_path.to_string_lossy().as_ref()), - "{stderr}" - ); + assert!(!stderr.contains("cache/artifacts"), "{stderr}"); + assert!(stderr.contains("create_assets"), "{stderr}"); + assert!(stderr.contains("assets/shared/report.txt"), "{stderr}"); + assert!(stderr.contains("fabro artifact cp"), "{stderr}"); } #[test] diff --git a/lib/crates/fabro-cli/tests/it/cmd/support.rs b/lib/crates/fabro-cli/tests/it/cmd/support.rs index 94f5379ca..2f9a98713 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/support.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/support.rs @@ -323,12 +323,6 @@ include = ["assets/**"] ); let run = run_local_workflow(context, &workspace_dir, "run.toml"); - assert!( - run.run_dir - .join("cache/artifacts/files/retry_assets/retry_2/assets/retry/report.txt") - .exists(), - "setup_artifact_run should materialize retry_2 assets" - ); WorkspaceRunSetup { run, workspace_dir } } diff --git a/lib/crates/fabro-config/src/storage.rs b/lib/crates/fabro-config/src/storage.rs index 542a4f89e..a14cf8ba8 100644 --- a/lib/crates/fabro-config/src/storage.rs +++ b/lib/crates/fabro-config/src/storage.rs @@ -115,27 +115,9 @@ impl RunScratch { self.root.join("runtime") } - #[must_use] - pub fn artifact_cache_dir(&self) -> PathBuf { - self.root.join("cache").join("artifacts") - } - - #[must_use] - pub fn artifact_files_dir(&self) -> PathBuf { - self.artifact_cache_dir().join("files") - } - - #[must_use] - pub fn artifact_stage_dir(&self, node_slug: &str, attempt: u32) -> PathBuf { - self.artifact_files_dir() - .join(node_slug) - .join(format!("retry_{attempt}")) - } - pub fn create(&self) -> std::io::Result<()> { std::fs::create_dir_all(self.worktree_dir())?; std::fs::create_dir_all(self.runtime_dir())?; - std::fs::create_dir_all(self.artifact_files_dir())?; Ok(()) } @@ -215,30 +197,12 @@ mod tests { assert_eq!(scratch.worktree_dir(), scratch.root().join("worktree")); assert_eq!(scratch.runtime_dir(), scratch.root().join("runtime")); - assert_eq!( - scratch.artifact_cache_dir(), - scratch.root().join("cache").join("artifacts") - ); - assert_eq!( - scratch.artifact_files_dir(), - scratch.root().join("cache").join("artifacts").join("files") - ); - assert_eq!( - scratch.artifact_stage_dir("plan", 2), - scratch - .root() - .join("cache") - .join("artifacts") - .join("files") - .join("plan") - .join("retry_2") - ); scratch.create().unwrap(); assert!(scratch.root().exists()); assert!(scratch.worktree_dir().exists()); assert!(scratch.runtime_dir().exists()); - assert!(scratch.artifact_files_dir().exists()); + assert!(!scratch.root().join("cache").join("artifacts").exists()); scratch.remove().unwrap(); assert!(!scratch.root().exists()); } diff --git a/lib/crates/fabro-server/src/demo/mod.rs b/lib/crates/fabro-server/src/demo/mod.rs index 581abb5bb..60ff9c54a 100644 --- a/lib/crates/fabro-server/src/demo/mod.rs +++ b/lib/crates/fabro-server/src/demo/mod.rs @@ -3,6 +3,7 @@ #![allow(clippy::default_trait_access, clippy::unreadable_literal)] use std::sync::Arc; +use std::time::Duration; use axum::Json; use axum::extract::{Path, Query, State}; @@ -188,7 +189,8 @@ pub(crate) async fn get_run_status( let elapsed_ms = item .timings .as_ref() - .map(|t| (t.elapsed_secs * 1000.0) as u64); + .and_then(|t| Duration::try_from_secs_f64(t.elapsed_secs).ok()) + .and_then(|duration| u64::try_from(duration.as_millis()).ok()); ( StatusCode::OK, Json(json!({ diff --git a/lib/crates/fabro-server/src/serve.rs b/lib/crates/fabro-server/src/serve.rs index 86d613ad2..9b092cd29 100644 --- a/lib/crates/fabro-server/src/serve.rs +++ b/lib/crates/fabro-server/src/serve.rs @@ -24,8 +24,8 @@ use crate::github_webhooks::WebhookManager; use crate::jwt_auth::{AuthMode, AuthStrategy, resolve_auth_mode_with_lookup}; use crate::secret_store::SecretStore; use crate::server::{ - build_app_state_with_path, reconcile_incomplete_runs_on_startup, shutdown_active_workers, - spawn_scheduler, + RouterOptions, build_app_state_with_path, build_router_with_options, + reconcile_incomplete_runs_on_startup, shutdown_active_workers, spawn_scheduler, }; use crate::tls::{ClientAuth, build_rustls_config, serve_tls_with_shutdown}; use fabro_llm::client::Client as LlmClient; @@ -266,8 +266,7 @@ where .expect("config lock poisoned") .web .as_ref() - .map(|web| web.enabled) - .unwrap_or(true); + .is_none_or(|web| web.enabled); let store_path = storage.store_dir(); let object_store = build_object_store(&store_path)?; @@ -299,11 +298,8 @@ where ); } spawn_scheduler(Arc::clone(&state)); - let router = crate::server::build_router_with_options( - Arc::clone(&state), - auth_mode, - crate::server::RouterOptions { web_enabled }, - ); + let router = + build_router_with_options(Arc::clone(&state), auth_mode, RouterOptions { web_enabled }); let bind_request = match args.bind { Some(ref s) => bind::parse_bind(s)?, diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index a9f77b7b3..94d158599 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -40,6 +40,7 @@ use fabro_types::{ }; use fabro_util::redact::redact_jsonl_line; use fabro_util::version::FABRO_VERSION; +use fabro_workflow::artifact_upload::ArtifactSink; use fabro_workflow::error::FabroError; use fabro_workflow::handler::HandlerRegistry; use futures_util::stream; @@ -850,9 +851,7 @@ pub fn build_router_with_options( debug!(method = %req.method(), path = %req.uri().path(), "HTTP request"); }) .on_response( - |response: &axum::response::Response, - latency: std::time::Duration, - _span: &tracing::Span| { + |response: &Response, latency: std::time::Duration, _span: &tracing::Span| { let status = response.status().as_u16(); let latency_ms = latency.as_millis(); if status >= 500 { @@ -3609,7 +3608,7 @@ async fn execute_run_in_process(state: Arc, run_id: RunId) { interviewer: Arc::clone(&interview_runtime), run_store: run_store.clone().into(), event_sink: workflow_event::RunEventSink::store(run_store.clone()), - artifact_uploader: None, + artifact_sink: Some(ArtifactSink::Store(state.artifact_store.clone())), run_control: None, github_app, on_node: None, diff --git a/lib/crates/fabro-server/src/static_files.rs b/lib/crates/fabro-server/src/static_files.rs index dc66bb77c..d83dd3bd9 100644 --- a/lib/crates/fabro-server/src/static_files.rs +++ b/lib/crates/fabro-server/src/static_files.rs @@ -96,7 +96,9 @@ fn has_hashed_extension(path: &str) -> bool { } fn is_source_map(path: &str) -> bool { - path.ends_with(".map") + Path::new(path) + .extension() + .is_some_and(|ext| ext.eq_ignore_ascii_case("map")) } #[cfg(test)] diff --git a/lib/crates/fabro-server/src/web_auth.rs b/lib/crates/fabro-server/src/web_auth.rs index 43f61d8b0..95ee3bb0f 100644 --- a/lib/crates/fabro-server/src/web_auth.rs +++ b/lib/crates/fabro-server/src/web_auth.rs @@ -560,7 +560,7 @@ async fn setup_register( settings.git = Some(git.clone()); if let Some(ref origin) = origin { let web = settings.web.get_or_insert_default(); - web.url = origin.to_string(); + web.url.clone_from(origin); } if let Some(parent) = settings_path.parent() { diff --git a/lib/crates/fabro-workflow/Cargo.toml b/lib/crates/fabro-workflow/Cargo.toml index d30ca8a78..729001040 100644 --- a/lib/crates/fabro-workflow/Cargo.toml +++ b/lib/crates/fabro-workflow/Cargo.toml @@ -61,13 +61,13 @@ tokio-util.workspace = true tracing.workspace = true walkdir.workspace = true reqwest.workspace = true +tempfile = "3" [dev-dependencies] base64.workspace = true toml.workspace = true fabro-mcp = { path = "../fabro-mcp" } tokio = { workspace = true, features = ["test-util", "macros"] } object_store.workspace = true -tempfile = "3" assert_cmd = "2" predicates = "3" fabro-macros = { path = "../fabro-macros" } diff --git a/lib/crates/fabro-workflow/src/artifact_upload.rs b/lib/crates/fabro-workflow/src/artifact_upload.rs index 7fffec8b0..93dd9cf90 100644 --- a/lib/crates/fabro-workflow/src/artifact_upload.rs +++ b/lib/crates/fabro-workflow/src/artifact_upload.rs @@ -1,7 +1,10 @@ +use std::sync::Arc; + use std::path::Path; use anyhow::Result; use async_trait::async_trait; +use fabro_store::ArtifactStore; use fabro_types::StageId; use crate::artifact_snapshot::CapturedArtifactInfo; @@ -15,3 +18,8 @@ pub trait StageArtifactUploader: Send + Sync { artifacts: &[CapturedArtifactInfo], ) -> Result<()>; } + +pub enum ArtifactSink { + Store(ArtifactStore), + Uploader(Arc), +} diff --git a/lib/crates/fabro-workflow/src/devcontainer_bridge.rs b/lib/crates/fabro-workflow/src/devcontainer_bridge.rs index 639b3042a..01763a187 100644 --- a/lib/crates/fabro-workflow/src/devcontainer_bridge.rs +++ b/lib/crates/fabro-workflow/src/devcontainer_bridge.rs @@ -202,8 +202,7 @@ async fn run_single_lifecycle_command( }); return Err(FabroError::engine(format!( "Devcontainer {phase} command failed (exit code {}): {command}\n{}", - result.exit_code, - result.stderr, + result.exit_code, result.stderr, ))); } emitter.emit(&Event::DevcontainerLifecycleCommandCompleted { @@ -499,15 +498,9 @@ mod tests { }); let sandbox = TestSandbox::with_exit_code(1); let commands = vec![fabro_devcontainer::Command::Shell("false".to_string())]; - let result = run_devcontainer_lifecycle( - &sandbox, - &emitter, - "on_create", - &commands, - 300_000, - None, - ) - .await; + let result = + run_devcontainer_lifecycle(&sandbox, &emitter, "on_create", &commands, 300_000, None) + .await; assert!(result.is_err()); let events = events.lock().unwrap(); assert!(events.iter().any(|event| { diff --git a/lib/crates/fabro-workflow/src/handler/manager_loop.rs b/lib/crates/fabro-workflow/src/handler/manager_loop.rs index a0742fc79..f8340c4ae 100644 --- a/lib/crates/fabro-workflow/src/handler/manager_loop.rs +++ b/lib/crates/fabro-workflow/src/handler/manager_loop.rs @@ -4,6 +4,7 @@ use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; +use crate::artifact_upload::ArtifactSink; use crate::condition::evaluate_condition; use crate::context::keys; use crate::context::{Context, WorkflowContext}; @@ -16,7 +17,7 @@ use crate::run_dir::visit_from_context; use crate::run_options::RunOptions; use async_trait::async_trait; use fabro_graphviz::graph::{AttrValue, Graph, Node}; -use fabro_store::Database; +use fabro_store::{ArtifactStore, Database}; use fabro_types::Settings; use object_store::memory::InMemory; use tokio::time::{sleep, timeout}; @@ -233,8 +234,9 @@ impl Handler for SubWorkflowHandler { let env = services.env.clone(); let dry_run = services.dry_run; let workflow_bundle = services.workflow_bundle.clone(); + let object_store = Arc::new(InMemory::new()); let store = Arc::new(Database::new( - Arc::new(InMemory::new()), + object_store.clone(), "", Duration::from_millis(1), )); @@ -242,6 +244,7 @@ impl Handler for SubWorkflowHandler { .create_run(&child_run_options.run_id) .await .map_err(|err| FabroError::engine(err.to_string()))?; + let artifact_store = ArtifactStore::new(object_store, "artifacts"); // Spawn child engine let mut child_handle = tokio::spawn(async move { @@ -258,7 +261,7 @@ impl Handler for SubWorkflowHandler { sandbox, registry, on_node: None, - artifact_uploader: None, + artifact_sink: Some(ArtifactSink::Store(artifact_store)), run_control: None, hook_runner, env, diff --git a/lib/crates/fabro-workflow/src/lifecycle/artifact.rs b/lib/crates/fabro-workflow/src/lifecycle/artifact.rs index 69c63685f..78e980dda 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/artifact.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/artifact.rs @@ -1,26 +1,26 @@ -use std::path::PathBuf; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Duration; use async_trait::async_trait; -use fabro_types::StageId; -use tokio::time::sleep; +use fabro_store::ArtifactStore; +use fabro_types::{RunId, StageId}; +use fabro_core::error::{CoreError, Result as CoreResult}; use fabro_core::graph::NodeSpec; use fabro_core::lifecycle::{AttemptContext, AttemptResultContext, RunLifecycle}; use fabro_core::outcome::NodeResult; use fabro_core::state::ExecutionState; +use tokio::{fs, time::sleep}; use crate::artifact::{normalize_durable_updates, offload_large_values, sync_artifacts_to_env}; use crate::artifact_snapshot::{CapturedArtifactInfo, collect_artifacts}; -use crate::artifact_upload::StageArtifactUploader; +use crate::artifact_upload::ArtifactSink; use crate::event::{Emitter, Event, RunNoticeLevel}; use crate::graph::WorkflowGraph; use crate::graph::WorkflowNode; use crate::outcome::BilledModelUsage; use crate::runtime_store::RunStoreHandle; -use fabro_core::error::Result as CoreResult; use fabro_core::lifecycle::NodeDecision; type WfRunState = ExecutionState>; @@ -38,9 +38,9 @@ pub(crate) struct ArtifactLifecycle { pub sandbox: Arc, pub run_store: RunStoreHandle, pub emitter: Arc, - pub artifacts_dir: PathBuf, + pub run_id: RunId, pub artifact_globs: Vec, - pub artifact_uploader: Option>, + pub artifact_sink: Option, pub captured_artifact_count: Arc, /// Per-attempt state: epoch seconds when the attempt started. attempt_start_epoch: std::sync::Mutex>, @@ -52,18 +52,18 @@ impl ArtifactLifecycle { sandbox: Arc, run_store: RunStoreHandle, emitter: Arc, - artifacts_dir: PathBuf, + run_id: RunId, artifact_globs: Vec, - artifact_uploader: Option>, + artifact_sink: Option, captured_artifact_count: Arc, ) -> Self { Self { sandbox, run_store, emitter, - artifacts_dir, + run_id, artifact_globs, - artifact_uploader, + artifact_sink, captured_artifact_count, attempt_start_epoch: std::sync::Mutex::new(None), } @@ -108,15 +108,12 @@ impl RunLifecycle for ArtifactLifecycle { } else { format!("{node_id}-visit_{visit}") }; - let artifact_capture_dir = self - .artifacts_dir - .join(&node_slug) - .join(format!("retry_{}", ctx.attempt)); - let _ = std::fs::create_dir_all(&artifact_capture_dir); + let artifact_capture_dir = + tempfile::tempdir().map_err(|err| CoreError::Other(err.to_string()))?; match collect_artifacts( &*self.sandbox, - &artifact_capture_dir, + artifact_capture_dir.path(), &self.artifact_globs, epoch, ) @@ -125,7 +122,11 @@ impl RunLifecycle for ArtifactLifecycle { Ok(summary) if summary.files_copied > 0 => { let stage_id = StageId::new(node_id.to_string(), ctx.attempt); if let Err(err) = self - .upload_artifacts(&stage_id, &artifact_capture_dir, &summary.captured_assets) + .persist_artifacts( + &stage_id, + artifact_capture_dir.path(), + &summary.captured_assets, + ) .await { self.emitter.emit(&Event::RunNotice { @@ -199,24 +200,24 @@ impl RunLifecycle for ArtifactLifecycle { } impl ArtifactLifecycle { - async fn upload_artifacts( + async fn persist_artifacts( &self, stage_id: &StageId, artifact_capture_dir: &std::path::Path, artifacts: &[CapturedArtifactInfo], ) -> Result<(), String> { - let Some(uploader) = self.artifact_uploader.as_ref() else { - return Ok(()); + let Some(sink) = self.artifact_sink.as_ref() else { + return Err("artifact sink is not configured".to_string()); }; let mut last_error = None; for attempt in 0..=ARTIFACT_UPLOAD_RETRY_DELAYS.len() { - match uploader - .upload_stage_artifacts(stage_id, artifact_capture_dir, artifacts) + match self + .persist_artifacts_once(sink, stage_id, artifact_capture_dir, artifacts) .await { Ok(()) => return Ok(()), - Err(err) => last_error = Some(err.to_string()), + Err(err) => last_error = Some(err), } if let Some(delay) = ARTIFACT_UPLOAD_RETRY_DELAYS.get(attempt) { @@ -226,4 +227,43 @@ impl ArtifactLifecycle { Err(last_error.unwrap_or_else(|| "artifact upload failed".to_string())) } + + async fn persist_artifacts_once( + &self, + sink: &ArtifactSink, + stage_id: &StageId, + artifact_capture_dir: &std::path::Path, + artifacts: &[CapturedArtifactInfo], + ) -> Result<(), String> { + match sink { + ArtifactSink::Store(store) => { + self.store_artifacts(store, stage_id, artifact_capture_dir, artifacts) + .await + } + ArtifactSink::Uploader(uploader) => uploader + .upload_stage_artifacts(stage_id, artifact_capture_dir, artifacts) + .await + .map_err(|err| err.to_string()), + } + } + + async fn store_artifacts( + &self, + store: &ArtifactStore, + stage_id: &StageId, + artifact_capture_dir: &std::path::Path, + artifacts: &[CapturedArtifactInfo], + ) -> Result<(), String> { + for artifact in artifacts { + let local_path = artifact_capture_dir.join(&artifact.path); + let bytes = fs::read(&local_path).await.map_err(|err| { + format!("failed to read artifact {}: {err}", local_path.display()) + })?; + store + .put(&self.run_id, stage_id, &artifact.path, &bytes) + .await + .map_err(|err| err.to_string())?; + } + Ok(()) + } } diff --git a/lib/crates/fabro-workflow/src/lifecycle/mod.rs b/lib/crates/fabro-workflow/src/lifecycle/mod.rs index 5457660c9..7e0106e79 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/mod.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/mod.rs @@ -7,13 +7,12 @@ pub(crate) mod git; pub(crate) mod hook; use std::collections::HashMap; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Instant; use async_trait::async_trait; -use fabro_config::RunScratch; use fabro_types::RunId; use fabro_core::error::Result as CoreResult; @@ -24,7 +23,7 @@ use fabro_core::lifecycle::{ use fabro_core::outcome::NodeResult; use fabro_core::state::ExecutionState; -use crate::artifact_upload::StageArtifactUploader; +use crate::artifact_upload::ArtifactSink; use crate::context; use crate::error::{FailureSignature, FailureSignatureExt}; use crate::event::Emitter; @@ -84,15 +83,14 @@ impl WorkflowLifecycle { hook_runner: Option>, sandbox: &Arc, graph: Arc, - run_dir: &PathBuf, + run_dir: &Path, run_store: &RunStoreHandle, - artifact_uploader: Option>, + artifact_sink: Option, run_options: &Arc, is_resume: bool, on_node: crate::OnNodeCallback, run_control: Option>, ) -> Self { - let run_scratch = RunScratch::new(run_dir); let restarted_from: Arc>> = Arc::new(Mutex::new(None)); let loop_restart_signature_limit = graph.loop_restart_signature_limit(); let checkpoint_git_result: Arc>> = @@ -145,7 +143,7 @@ impl WorkflowLifecycle { Arc::clone(&graph), Arc::clone(sandbox), run_store.clone(), - run_dir.clone(), + run_dir.to_path_buf(), ); let start_node_id = graph.find_start_node().map(|n| n.id.clone()); @@ -166,9 +164,9 @@ impl WorkflowLifecycle { Arc::clone(sandbox), run_store.clone(), Arc::clone(emitter), - run_scratch.artifact_files_dir(), + run_options.run_id, run_options.artifact_globs().to_vec(), - artifact_uploader, + artifact_sink, captured_artifact_count, ); diff --git a/lib/crates/fabro-workflow/src/node_handler.rs b/lib/crates/fabro-workflow/src/node_handler.rs index 3a7952f57..5e85d9bf8 100644 --- a/lib/crates/fabro-workflow/src/node_handler.rs +++ b/lib/crates/fabro-workflow/src/node_handler.rs @@ -12,6 +12,7 @@ use fabro_core::retry::RetryPolicy as CoreRetryPolicy; use crate::artifact; use crate::context::Context; +use crate::error::FabroError; use crate::graph::WorkflowGraph; use crate::graph::WorkflowNode; @@ -103,7 +104,7 @@ impl NodeHandler for WorkflowNodeHandler { match timed_result { Ok(Ok(wf_outcome)) => Ok(wf_outcome), - Ok(Err(crate::error::FabroError::Cancelled)) => Err(CoreError::Cancelled), + Ok(Err(FabroError::Cancelled)) => Err(CoreError::Cancelled), Ok(Err(fabro_err)) => { let retryable = handler.should_retry(&fabro_err); Err(CoreError::handler(HandlerErrorDetail { diff --git a/lib/crates/fabro-workflow/src/operations/start.rs b/lib/crates/fabro-workflow/src/operations/start.rs index e749a0e27..40bef59d0 100644 --- a/lib/crates/fabro-workflow/src/operations/start.rs +++ b/lib/crates/fabro-workflow/src/operations/start.rs @@ -11,7 +11,7 @@ use fabro_model::{Catalog, FallbackTarget, Provider}; use fabro_sandbox::{SandboxProvider, SandboxSpec}; use fabro_types::{RunId, Settings}; -use crate::artifact_upload::StageArtifactUploader; +use crate::artifact_upload::ArtifactSink; use crate::context::Context; use crate::error::FabroError; use crate::event::{ @@ -51,7 +51,7 @@ struct RunSession { seed_context: Option, run_store: RunStoreHandle, event_sink: RunEventSink, - artifact_uploader: Option>, + artifact_sink: Option, git: Option, github_app: Option, worktree_mode: Option, @@ -74,7 +74,7 @@ pub struct StartServices { pub interviewer: Arc, pub run_store: RunStoreHandle, pub event_sink: RunEventSink, - pub artifact_uploader: Option>, + pub artifact_sink: Option, pub run_control: Option>, pub github_app: Option, pub on_node: crate::OnNodeCallback, @@ -402,7 +402,7 @@ impl RunSession { devcontainer, seed_context: None, run_store: services.run_store, - artifact_uploader: services.artifact_uploader, + artifact_sink: services.artifact_sink, git, github_app: services.github_app.clone(), worktree_mode: Some(resolve_worktree_mode(&settings)), @@ -549,7 +549,7 @@ impl RunSession { git: self.git, worktree_mode: self.worktree_mode, registry_override: self.registry_override, - artifact_uploader: self.artifact_uploader, + artifact_sink: self.artifact_sink, run_control: self.run_control, checkpoint, seed_context: self.seed_context, @@ -900,7 +900,7 @@ mod tests { interviewer: Arc::new(fabro_interview::AutoApproveInterviewer), run_store: store.open_run(&fixtures::RUN_1).await.unwrap().into(), event_sink: RunEventSink::store(store.open_run(&fixtures::RUN_1).await.unwrap()), - artifact_uploader: None, + artifact_sink: None, run_control: None, github_app: None, on_node: None, diff --git a/lib/crates/fabro-workflow/src/pipeline/execute.rs b/lib/crates/fabro-workflow/src/pipeline/execute.rs index 3e68c5977..902228342 100644 --- a/lib/crates/fabro-workflow/src/pipeline/execute.rs +++ b/lib/crates/fabro-workflow/src/pipeline/execute.rs @@ -48,7 +48,7 @@ pub async fn execute(init: Initialized) -> Executed { sandbox, registry, on_node, - artifact_uploader, + artifact_sink, run_control, hook_runner, env, @@ -107,7 +107,7 @@ pub async fn execute(init: Initialized) -> Executed { graph_arc, &run_options.run_dir, &run_store, - artifact_uploader, + artifact_sink, &settings_arc, checkpoint.is_some(), on_node, diff --git a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs index ecc2bca1c..b88ef269a 100644 --- a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs +++ b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs @@ -222,7 +222,7 @@ async fn execute_test_run_with_options( worktree_mode: None, run_control: None, registry_override, - artifact_uploader: None, + artifact_sink: None, checkpoint: None, seed_context: None, }, @@ -279,7 +279,7 @@ async fn execute_runs_start_to_exit_and_returns_final_context() { worktree_mode: None, run_control: None, registry_override: None, - artifact_uploader: None, + artifact_sink: None, checkpoint: None, seed_context: None, }, @@ -346,7 +346,7 @@ async fn run_with_lifecycle( worktree_mode: None, run_control: None, registry_override: Some(Arc::new(registry)), - artifact_uploader: None, + artifact_sink: None, checkpoint: None, seed_context: None, }, diff --git a/lib/crates/fabro-workflow/src/pipeline/initialize.rs b/lib/crates/fabro-workflow/src/pipeline/initialize.rs index 2e182d826..2b8517de3 100644 --- a/lib/crates/fabro-workflow/src/pipeline/initialize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/initialize.rs @@ -19,7 +19,7 @@ use crate::error::FabroError; use crate::event::{Emitter, Event, RunNoticeLevel}; use crate::git::{self, GitSyncStatus, MetadataStore}; use crate::handler::llm::{AgentApiBackend, AgentCliBackend, BackendRouter}; -use crate::handler::{HandlerRegistry, default_registry}; +use crate::handler::{HandlerRegistry, default_registry, sandbox_cancel_token}; use crate::run_options::GitCheckpointOptions; use tokio::process::Command as TokioCommand; use tokio::runtime::Handle; @@ -589,8 +589,7 @@ pub async fn initialize( index, }); let cmd_start = Instant::now(); - let cancel_token = - crate::handler::sandbox_cancel_token(options.run_options.cancel_token.clone()); + let cancel_token = sandbox_cancel_token(options.run_options.cancel_token.clone()); let result = sandbox .exec_command( command, @@ -659,7 +658,7 @@ pub async fn initialize( sandbox, registry, on_node: None, - artifact_uploader: options.artifact_uploader, + artifact_sink: options.artifact_sink, run_control: options.run_control, hook_runner, env, @@ -816,7 +815,7 @@ mod tests { worktree_mode: None, run_control: None, registry_override: None, - artifact_uploader: None, + artifact_sink: None, checkpoint: None, seed_context: None, }, @@ -893,7 +892,7 @@ mod tests { worktree_mode: None, run_control: None, registry_override: None, - artifact_uploader: None, + artifact_sink: None, checkpoint: None, seed_context: None, }, @@ -964,7 +963,7 @@ mod tests { worktree_mode: None, run_control: None, registry_override: None, - artifact_uploader: None, + artifact_sink: None, checkpoint: None, seed_context: None, }, @@ -1030,7 +1029,7 @@ mod tests { worktree_mode: None, run_control: None, registry_override: None, - artifact_uploader: None, + artifact_sink: None, checkpoint: None, seed_context: None, }, diff --git a/lib/crates/fabro-workflow/src/pipeline/types.rs b/lib/crates/fabro-workflow/src/pipeline/types.rs index b3605429a..c2ae831aa 100644 --- a/lib/crates/fabro-workflow/src/pipeline/types.rs +++ b/lib/crates/fabro-workflow/src/pipeline/types.rs @@ -14,7 +14,7 @@ use fabro_sandbox::SandboxSpec; use fabro_types::RunId; use fabro_validate::Diagnostic; -use crate::artifact_upload::StageArtifactUploader; +use crate::artifact_upload::ArtifactSink; use crate::context::Context; use crate::error::FabroError; use crate::event::Emitter; @@ -248,7 +248,7 @@ pub struct InitOptions { pub git: Option, pub worktree_mode: Option, pub registry_override: Option>, - pub artifact_uploader: Option>, + pub artifact_sink: Option, pub run_control: Option>, pub checkpoint: Option, pub seed_context: Option, @@ -269,7 +269,7 @@ pub struct Initialized { pub sandbox: Arc, pub registry: Arc, pub on_node: crate::OnNodeCallback, - pub artifact_uploader: Option>, + pub artifact_sink: Option, pub run_control: Option>, pub hook_runner: Option>, pub env: HashMap, diff --git a/lib/crates/fabro-workflow/src/test_support.rs b/lib/crates/fabro-workflow/src/test_support.rs index 7c6d68fa6..ebe61f89d 100644 --- a/lib/crates/fabro-workflow/src/test_support.rs +++ b/lib/crates/fabro-workflow/src/test_support.rs @@ -6,9 +6,10 @@ use std::time::Duration; use fabro_agent::Sandbox; use fabro_graphviz::graph::Graph as GvGraph; -use fabro_store::{Database, RunProjection}; +use fabro_store::{ArtifactStore, Database, RunProjection}; use object_store::local::LocalFileSystem; +use crate::artifact_upload::ArtifactSink; use crate::error::{FabroError, Result}; use crate::event::{Emitter, Event, StoreProgressLogger, append_event}; use crate::handler::HandlerRegistry; @@ -106,6 +107,13 @@ async fn initialized( let emitter = bound_emitter(run_options.run_id, &emitter); let store_logger = StoreProgressLogger::new(run_store.clone()); store_logger.register(emitter.as_ref()); + let artifact_store = ArtifactStore::new( + Arc::new( + LocalFileSystem::new_with_prefix(&store_dir) + .expect("failed to create local test artifact store"), + ), + "artifacts", + ); InitializedState { initialized: Initialized { graph: graph.clone(), @@ -120,7 +128,7 @@ async fn initialized( sandbox, registry: Arc::new(registry), on_node: None, - artifact_uploader: None, + artifact_sink: Some(ArtifactSink::Store(artifact_store)), run_control: None, hook_runner: options.hook_runner, env: options.env, diff --git a/lib/crates/fabro-workflow/tests/it/daytona_integration.rs b/lib/crates/fabro-workflow/tests/it/daytona_integration.rs index 813fefecc..514a38eca 100644 --- a/lib/crates/fabro-workflow/tests/it/daytona_integration.rs +++ b/lib/crates/fabro-workflow/tests/it/daytona_integration.rs @@ -18,12 +18,11 @@ use std::path::Path; use std::sync::Arc; use fabro_agent::Sandbox; -use fabro_config::RunScratch; use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node}; use fabro_llm::provider::Provider; use fabro_sandbox::daytona::{DaytonaConfig, DaytonaSandbox, DaytonaSnapshotConfig}; -use fabro_store::Database; -use fabro_types::{RunId, Settings}; +use fabro_store::{ArtifactStore, Database}; +use fabro_types::{RunId, Settings, StageId}; use fabro_workflow::artifact::sync_artifacts_to_env; use fabro_workflow::context::Context; use fabro_workflow::error::FabroError; @@ -146,6 +145,14 @@ async fn create_env() -> DaytonaSandbox { create_env_with_github_app(Some(creds)).await } +fn test_artifact_store(run_dir: &Path) -> ArtifactStore { + let object_store = Arc::new( + LocalFileSystem::new_with_prefix(test_store_dir(run_dir)) + .expect("failed to create local artifact store"), + ); + ArtifactStore::new(object_store, "artifacts") +} + async fn create_env_with_github_app( github_app: Option, ) -> DaytonaSandbox { @@ -1342,16 +1349,24 @@ async fn daytona_asset_collection() { .expect("pipeline should succeed"); assert_eq!(outcome.status, StageStatus::Success); - let artifacts_dir = RunScratch::new(dir.path()).artifact_stage_dir("create_assets", 1); - - let report_path = artifacts_dir.join("test-results/report.xml"); - assert!( - report_path.exists(), - "report.xml should be collected from Daytona sandbox at {}", - report_path.display() - ); - let content = std::fs::read_to_string(&report_path).unwrap(); + let content = String::from_utf8( + test_artifact_store(dir.path()) + .get( + &run_options.run_id, + &StageId::new("create_assets", 1), + "test-results/report.xml", + ) + .await + .unwrap() + .expect("artifact should be stored from Daytona sandbox") + .to_vec(), + ) + .unwrap(); assert!(content.contains("testsuites")); + assert!( + !dir.path().join("cache").join("artifacts").exists(), + "artifact scratch cache should not be created" + ); env.cleanup().await.unwrap(); } diff --git a/lib/crates/fabro-workflow/tests/it/integration.rs b/lib/crates/fabro-workflow/tests/it/integration.rs index c17d4a0c2..6fd54edf6 100644 --- a/lib/crates/fabro-workflow/tests/it/integration.rs +++ b/lib/crates/fabro-workflow/tests/it/integration.rs @@ -25,8 +25,8 @@ use fabro_interview::{ QueueInterviewer, RecordingInterviewer, }; use fabro_llm::provider::Provider; -use fabro_store::Database; -use fabro_types::{RunEvent, RunId, Settings}; +use fabro_store::{ArtifactStore, Database}; +use fabro_types::{RunEvent, RunId, Settings, StageId}; use fabro_validate::{Severity, validate, validate_or_raise}; use fabro_workflow::context::Context; use fabro_workflow::error::{FabroError, FailureSignatureExt}; @@ -176,6 +176,14 @@ fn save_checkpoint(path: &Path, checkpoint: &Checkpoint) { std::fs::write(path, serde_json::to_string_pretty(checkpoint).unwrap()).unwrap(); } +fn test_artifact_store(run_dir: &Path) -> ArtifactStore { + let object_store = Arc::new( + LocalFileSystem::new_with_prefix(test_store_dir(run_dir)) + .expect("failed to create local artifact store"), + ); + ArtifactStore::new(object_store, "artifacts") +} + // --------------------------------------------------------------------------- // 1. Parse and validate all 3 spec examples (Section 2.13) // --------------------------------------------------------------------------- @@ -12639,17 +12647,38 @@ async fn asset_collection_local_sandbox_success() { .expect("run should succeed"); assert_eq!(outcome.status, StageStatus::Success); - // Check that artifact files were collected into the stage directory - let artifacts_dir = RunScratch::new(run_dir.path()).artifact_stage_dir("create_assets", 1); - - let report_path = artifacts_dir.join("test-results/report.xml"); - assert!( - report_path.exists(), - "report.xml should be collected at {}", - report_path.display() + let artifact_store = test_artifact_store(run_dir.path()); + let artifacts = artifact_store + .list_for_run(&run_options.run_id) + .await + .unwrap(); + assert_eq!( + artifacts.len(), + 2, + "expected stored artifacts for both files" ); - let report_content = std::fs::read_to_string(&report_path).unwrap(); + assert_eq!(artifacts[0].node, StageId::new("create_assets", 1)); + assert_eq!(artifacts[0].filename, "test-results/output.txt"); + assert_eq!(artifacts[1].node, StageId::new("create_assets", 1)); + assert_eq!(artifacts[1].filename, "test-results/report.xml"); + let report_content = String::from_utf8( + artifact_store + .get( + &run_options.run_id, + &StageId::new("create_assets", 1), + "test-results/report.xml", + ) + .await + .unwrap() + .expect("artifact should be stored") + .to_vec(), + ) + .unwrap(); assert!(report_content.contains("testsuites")); + assert!( + !run_dir.path().join("cache").join("artifacts").exists(), + "artifact scratch cache should not be created" + ); // Check that ArtifactCaptured events were emitted let captured_events = events.lock().unwrap(); @@ -12749,13 +12778,23 @@ async fn asset_collection_local_sandbox_on_failure() { // Assets should still be collected regardless of intermediate node failures. assert_eq!(outcome.status, StageStatus::Success); - let artifacts_dir = RunScratch::new(run_dir.path()).artifact_stage_dir("create_assets", 1); - - let report_path = artifacts_dir.join("test-results/report.xml"); + let report_content = String::from_utf8( + test_artifact_store(run_dir.path()) + .get( + &run_options.run_id, + &StageId::new("create_assets", 1), + "test-results/report.xml", + ) + .await + .unwrap() + .expect("artifact should still be stored after handler failure") + .to_vec(), + ) + .unwrap(); + assert!(report_content.contains("testsuites")); assert!( - report_path.exists(), - "report.xml should still be collected after handler failure, at {}", - report_path.display() + !run_dir.path().join("cache").join("artifacts").exists(), + "artifact scratch cache should not be created" ); } @@ -12838,16 +12877,24 @@ async fn asset_collection_docker_sandbox() { .expect("pipeline should succeed"); assert_eq!(outcome.status, StageStatus::Success); - let artifacts_dir = RunScratch::new(run_dir.path()).artifact_stage_dir("create_assets", 1); - - let report_path = artifacts_dir.join("test-results/report.xml"); - assert!( - report_path.exists(), - "report.xml should be collected from Docker container at {}", - report_path.display() - ); - let content = std::fs::read_to_string(&report_path).unwrap(); + let content = String::from_utf8( + test_artifact_store(run_dir.path()) + .get( + &run_options.run_id, + &StageId::new("create_assets", 1), + "test-results/report.xml", + ) + .await + .unwrap() + .expect("artifact should be stored from Docker container") + .to_vec(), + ) + .unwrap(); assert!(content.contains("testsuites")); + assert!( + !run_dir.path().join("cache").join("artifacts").exists(), + "artifact scratch cache should not be created" + ); sandbox.cleanup().await.unwrap(); }