Test blob offloads through production hydration

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<Checkpoint, Box<dyn std::error:
         .ok_or_else(|| "checkpoint should exist in run store".into())
 }
 
+async fn resolve_checkpoint_text(
+    run_dir: &Path,
+    run_id: &RunId,
+    value: &serde_json::Value,
+) -> Result<String, Box<dyn std::error::Error>> {
+    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();
 }

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::<String>(&bytes)?)
+            let run_store = RunStoreHandle::from(run);
+            Ok(runtime.block_on(artifact::resolve_text_or_blob_ref_str(&current, &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!(