diff --git a/lib/components/fabro-workflow/tests/it/daytona_integration.rs b/lib/components/fabro-workflow/tests/it/daytona_integration.rs index 0acfb0f60..9932f5a52 100644 --- a/lib/components/fabro-workflow/tests/it/daytona_integration.rs +++ b/lib/components/fabro-workflow/tests/it/daytona_integration.rs @@ -27,9 +27,9 @@ use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node}; use fabro_sandbox::daytona::{DaytonaConfig, DaytonaSandbox}; use fabro_static::EnvVars; use fabro_store::{ArtifactKey, ArtifactStore, Database}; -use fabro_types::{RunId, StageId, WorkflowSettings}; +use fabro_types::{RunId, StageId, WorkflowSettings, parse_blob_ref}; use fabro_util::shell; -use fabro_workflow::artifact::sync_artifacts_to_env; +use fabro_workflow::artifact; use fabro_workflow::context::Context; use fabro_workflow::error::Error; use fabro_workflow::event::Emitter; @@ -39,6 +39,7 @@ use fabro_workflow::handler::{Handler, HandlerRegistry}; use fabro_workflow::outcome::{Outcome, StageOutcome}; use fabro_workflow::records::Checkpoint; use fabro_workflow::run_options::{GitCheckpointOptions, RunOptions}; +use fabro_workflow::runtime_store::RunStoreHandle; use fabro_workflow::test_support::{WorkflowRunner, test_store_dir}; use object_store::local::LocalFileSystem; use tokio_util::sync::CancellationToken; @@ -159,6 +160,25 @@ fn load_run_checkpoint(run_dir: &Path) -> Result Result> { + let Some(current) = value.as_str() else { + return Ok(value.to_string()); + }; + if parse_blob_ref(current).is_none() { + return Ok(current.to_string()); + } + + let object_store = Arc::new(LocalFileSystem::new_with_prefix(test_store_dir(run_dir))?); + let store = Database::new(object_store, "", std::time::Duration::from_millis(1), None); + let run = store.open_run_reader(run_id).await?; + let run_store = RunStoreHandle::from(run); + Ok(artifact::resolve_text_or_blob_ref_str(current, &run_store).await?) +} + async fn create_env() -> DaytonaSandbox { let creds = load_github_app_credentials(); create_env_with_github_app(Some(creds)).await @@ -419,7 +439,9 @@ async fn daytona_artifact_sync_uploads_and_rewrites_pointer() { // Sync — the local file doesn't exist in the Daytona sandbox, so it should // upload - sync_artifacts_to_env(&mut updates, &env).await.unwrap(); + artifact::sync_artifacts_to_env(&mut updates, &env) + .await + .unwrap(); // Pointer should be rewritten to the Daytona working directory let new_pointer = updates["response.plan"].as_str().unwrap(); @@ -544,15 +566,18 @@ async fn daytona_pipeline_artifact_offload_and_sync() { .get("response.big_output") .expect("context should have response.big_output"); let pointer_str = pointer_value.as_str().expect("pointer should be a string"); - let expected_blob_hash = fabro_types::BlobHash::new( - &serde_json::to_vec(&serde_json::json!("x".repeat(150 * 1024))) - .expect("large value should serialize"), - ); - assert_eq!( - pointer_str, - fabro_types::format_blob_ref(&expected_blob_hash), + assert!( + parse_blob_ref(pointer_str).is_some(), "checkpoint should persist a blob ref" ); + let resolved = resolve_checkpoint_text(dir.path(), &run_options.run_id, pointer_value) + .await + .expect("offloaded value should resolve through the run store"); + assert_eq!( + resolved, + "x".repeat(150 * 1024), + "offloaded value should round-trip through the run store" + ); env.cleanup().await.unwrap(); } diff --git a/lib/components/fabro-workflow/tests/it/integration.rs b/lib/components/fabro-workflow/tests/it/integration.rs index 02e0e74fb..42ea14630 100644 --- a/lib/components/fabro-workflow/tests/it/integration.rs +++ b/lib/components/fabro-workflow/tests/it/integration.rs @@ -35,6 +35,7 @@ use fabro_model::{Catalog, ProviderId}; use fabro_store::{ArtifactKey, ArtifactStore, Database}; use fabro_types::{EventBody, RunEvent, RunId, StageId, WorkflowSettings, parse_blob_ref}; use fabro_validate::{Severity, validate, validate_or_raise}; +use fabro_workflow::artifact; use fabro_workflow::context::Context; use fabro_workflow::error::{Error, FailureSignatureExt}; use fabro_workflow::event::{Emitter, Event}; @@ -54,6 +55,7 @@ use fabro_workflow::model_fallback::ModelFallbackPolicy; use fabro_workflow::outcome::{Outcome, OutcomeExt, StageOutcome}; use fabro_workflow::records::{Checkpoint, CheckpointExt}; use fabro_workflow::run_options::{GitCheckpointOptions, RunOptions}; +use fabro_workflow::runtime_store::RunStoreHandle; use fabro_workflow::test_support::{ WorkflowRunner, collect_events, run_graph_with_hooks, test_store_dir, }; @@ -233,10 +235,11 @@ fn resolve_checkpoint_text( let Some(current) = value.as_str() else { return Ok(value.to_string()); }; - let Some(blob_hash) = parse_blob_ref(current) else { + if parse_blob_ref(current).is_none() { return Ok(current.to_string()); - }; + } + let current = current.to_string(); let run_dir = run_dir.to_path_buf(); let (store_dir, uses_shared_store) = run_store_dir_and_mode(&run_dir)?; std::thread::spawn( @@ -271,10 +274,8 @@ fn resolve_checkpoint_text( .id }; let run = runtime.block_on(store.open_run_reader(&run_id))?; - let bytes = runtime - .block_on(run.read_blob(&blob_hash))? - .ok_or("checkpoint blob should exist")?; - Ok(serde_json::from_slice::(&bytes)?) + let run_store = RunStoreHandle::from(run); + Ok(runtime.block_on(artifact::resolve_text_or_blob_ref_str(¤t, &run_store))?) }, ) .join() @@ -10059,15 +10060,17 @@ 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"); - let expected_blob_hash = fabro_types::BlobHash::new( - &serde_json::to_vec(&serde_json::json!("x".repeat(150 * 1024))) - .expect("large value should serialize"), - ); - assert_eq!( - pointer_str, - fabro_types::format_blob_ref(&expected_blob_hash), + assert!( + parse_blob_ref(pointer_str).is_some(), "value should be a durable blob ref" ); + let resolved = resolve_checkpoint_text(dir.path(), pointer_value) + .expect("offloaded value should resolve through the run store"); + assert_eq!( + resolved, + "x".repeat(150 * 1024), + "offloaded value should round-trip through the run store" + ); // WorkflowRunCompleted artifact_count now tracks captured artifacts, not // offloaded values. @@ -10258,15 +10261,17 @@ async fn artifact_pointers_rewritten_for_remote_sandbox() { .get("response.big_output") .expect("context should have response.big_output"); let pointer_str = pointer_value.as_str().expect("pointer should be a string"); - let expected_blob_hash = fabro_types::BlobHash::new( - &serde_json::to_vec(&serde_json::json!("x".repeat(150 * 1024))) - .expect("large value should serialize"), - ); - assert_eq!( - pointer_str, - fabro_types::format_blob_ref(&expected_blob_hash), + assert!( + parse_blob_ref(pointer_str).is_some(), "checkpoint should persist a blob ref" ); + let resolved = resolve_checkpoint_text(dir.path(), pointer_value) + .expect("offloaded value should resolve through the run store"); + assert_eq!( + resolved, + "x".repeat(150 * 1024), + "offloaded value should round-trip through the run store" + ); let written = remote_env.written.lock().unwrap(); assert!(